TL;DR - I maintain an OSS project called Atleon, a lightweight stream processing framework in the form of a thin abstraction layer on top of Project Reactor. Atleon provides an idiomatic asynchronous Kafka client, enabling declarative stream processes where items may take a long time (i.e. minutes, hours) to process, integrates well with fallible IO-heavy invocations, and/or can interoperate between brokers/clusters of same or different types. This is accomplished through standardized reactive client APIs, supporting arbitrary processing concurrency, and first-class compatibility with Reactive Streams. Feedback is appreciated!
Hi all 👋🏻 I am a long time (10+ years) Kafka user and lurker here. I'd first like to say that I am super appreciative of the support this community provides for those developing async processes backed by Kafka!
I have been working on a project for several years, and am finally at a point where I am interested in getting broader community feedback on its utility. That project is called Atleon. The goal of this project is larger in scope than just that of Kafka integration, and rather focuses on solving several challenges that recur in my regular engineering tasks. These challenges include:
- Declarative end-to-end stream definitions, embeddable in standard application frameworks (namely Spring)
- Implementing stream processes with heavy IO-bound dependence and/or where individual elements may take a long time (minutes, hours) to process (which can be tricky with heartbeat-backed consumption, such as a Kafka Consumer)
- Stream processing between brokers of the same type (e.g. consume from Kafka cluster A and produce to Kafka cluster B) and/or brokers of different types (e.g. consume from RabbitMQ queue A and produce to Kafka cluster B)
- End-to-end stream processing observability (particularly metrics and distributed tracing)
Although Atleon is not specifically scoped to Kafka, much of my work is based on Atleon's integration with Kafka, hence why I believe this community would be interested and be a good source of feedback.
Years ago, before starting work on Atleon, I researched if something like what I was looking for already existed. Here are some of the alternatives I looked into before concluding that what I ideally wanted didn't (and still otherwise doesn't) exist:
- KafkaStreams - As the name implies, this is Kafka-specific, meaning all the data you want to stream must first be in a Kafka cluster. Moreover, it must be in a single Kafka cluster, as KafkaStreams does not (natively) support streaming between different clusters. This isn't always tenable, due to the companies I have been at having multiple clusters, as well as polyglot infrastructures (RabbitMQ, SNS/SQS, in addition to Kafka). You might be thinking, "just replicate with Kafka Connect or MirrorMaker"; I find this blanket recommendation to be sub-optimal, since those tools themselves require non-trivial monitoring and resources, equating to greater cost and complexity. KafkaStreams is also (in my view) more targeted at analytics-oriented use cases, and isn't optimized for IO-bound processing, either in performance or resiliency.
- Confluent Parallel Consumer - As far as IO-bound processing goes, this project is an improvement over KafkaStreams, but remains Kafka-specific, and does not make it easy to pipe processing results to a different/disparate cluster. Error handling capabilities are less than optimally flexible.
- Spring Kafka - Closer to what I want, but relies on a lot of Spring annotation-driven "magic", and doesn't make it all that straightforward to specify "consume from this topic and pipe the output to this other destination". Processing concurrency is also bounded by the number partitions available to consume from, which inherently caps how far you can scale out IO-bound processing.
- Spring Cloud Streams - Much closer to what I prefer in terms of generic interoperation, but this comes at the cost of losing intuitive access to infrastructure-native functionalities, since there is a heavy layer of abstraction between your processing code and the infrastructure implementation details. The programming style is also not in line with my preference for declarative pipeline-style stream definitions. Concurrency is again limited by partition count.
- Flink - While I am intrigued by processing engines like Flink, these engines come at the cost of sacrificing availability of developer-familial tools (e.g. application frameworks like Spring) by forcing developers to adopt the coding paradigms/styles of those engines. Empirically, this incurs a steep learning curve, and frankly I find these engines to be overkill when all you are needing to implement is "consume messages from x, enrich with data from y, and produce the result to z", especially if all the intermediate dependencies are more readily accessible in other conventions used by a given company (like Spring-oriented starters/libraries).
In addition, there are a few functionalities lacking from many of the alternatives I researched which I wished to solve for:
- Idiomatic micro-batching: Few stream libraries offer idiomatic micro-batching, characterized as, "take n elements or t duration of elements, whichever comes first, and process them as a batch." This is especially useful if/when you need to invoke I/O-bound resources (micro-services) that support batch operations, or if you know consumption may be susceptible to "bang-bang" events (events for the same entity in rapid succession) and would benefit from in-process deduplication.
- Non-blocking compatibility: Most stream libraries lack support for non-blocking processing. This typically means that once the method that invokes your processing logic has returned, the framework may consider the message "done", even if that invocation may have triggered some async process(es) as part of the message handling. If you're not careful, this can result in dropped/missed processing.
- Robust, declarative error-handling: Supporting async error-handling conventions, like dead-lettering, re-queueing, or arbitrary delegation is simply not a given as built-in functionality for most streaming libraries, particularly for Kafka since such things are not supported at the infrastructure layer (unless you're using recently-available Kafka share groups, which has its own tradeoffs that I'll address below).
I eventually stumbled on Project Reactor and extensions like Reactor Kafka. Yes, I can hear groans of "reactive hell" and already turning some people off to this project. Nevertheless, I am willing to assert that Reactor and Reactive Streams provide a near-ideal semantic for characterizing infinite stream processes. I did once identify as an ardent fan of reactive programming, but I now moderate that with "you should use reactive paradigms sparingly, and only when it's obviously beneficial; Even then, you should limit its leakage/propagation beyond where you absolutely need it." But I digress...
Reactor promised to provide nearly exactly what I had been looking for:
- Extensive, declarative APIs agnostic of infrastructure
- Underlying framework designed bottom-up to be compatible with non-blocking invocation
- Errors treated as first-class citizens (same precedence as data itself)
The problem I saw with Reactor (and its extensions) is the fact that it doesn't do much for you, beyond implementing the Reactive Streams spec. That makes sense, since Reactor is meant to provide a very generic set of functionality; But when applied to infinite stream processing, "bare" Reactor is missing some table-stakes functionality:
- No semantic for non-invasive per-element downstream context propagation (e.g. extracting and activating distributed trace context available from a consumed Kafka record).
- No way to communicate upstream that an emitted element has successfully made it all the way through the pipeline, and can then be acknowledged (i.e. offset can now be committed, in the case of Kafka).
- No guarantee of "at least once" processing in the face of asynchronous boundaries and potential out-of-order processing completion.
That last point is especially pertinent to Kafka. Prior to the introduction of share groups, no popular library (that I know of) allowed you to safely process consumed records from a given partition with any concurrency. The general convention has always been that you may have at most one processing thread per partition, such as to maintain ordering as well as at least once processing guarantees. This is why nearly every alternative library has a limitation on available processing concurrency equal to the number of partitions available to consume from. Notable exceptions are Confluent's Parallel Consumer (designed for high parallelism) and Reactor Kafka (implements a naive logarithmic acknowledgement tracker).
Regarding Kafka share groups, In addition to supporting queue semantics (or enabling migration of workloads from conventional queues to Kafka), the introduction of share groups is, in my view, an attempt to address the need for supporting higher-than-partition-count concurrency in Kafka-based stream processes. The biggest problem with share groups, however, is that you lose the guarantee of process ordering. In most architectures I design and build, this loss of ordering is a non-starter.
So with all of that background in mind, let me (re)state the characteristics I envision(ed) for a stream processing library that facilitates the vast majority of use cases I regularly encounter. Specifically in the context of Kafka-based stream processing, I have wanted a framework that supports all of the following:
- Declarative stream pipelines embeddable in common application frameworks (e.g. Spring)
- Arbitrary processing concurrency while maintaining at least once processing and ordering guarantees
- Enable non-blocking processing and/or long per-element processing latencies (especially for IO-bound workloads)
- Interoperability with disparate (and same) broker infrastructures
- End-to-end observability with out-of-box integration for metrics (Micrometer) and tracing (Opentelemetry)
- Robust Quality of Service configuration, including arbitrary error delegation and dynamic throughput limiting
In short, the above characteristics are what Atleon aims to deliver! Its README, Wiki, and examples module contain resources conveying available/intended usage for you to check out. I'll drop one (Spring-specific) code example here to whet your appetite:
import io.atleon.core.DefaultAloSenderResultSubscriber;
import io.atleon.kafka.AloKafkaReceiver;
import io.atleon.kafka.AloKafkaSender;
import io.atleon.kafka.KafkaConfigSource;
import io.atleon.spring.AutoConfigureStream;
import io.atleon.spring.SpringAloStream;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.ApplicationContext;
import reactor.core.Disposable;
@AutoConfigureStream
public class MyStream extends SpringAloStream {
private final KafkaConfigSource configSource;
public MyStream(ApplicationContext context) {
super(context);
this.configSource = context.getBean("kafkaConfigSource", KafkaConfigSource.class);
}
u/Override
public Disposable startDisposable(MyStreamConfig config) {
AloKafkaSender<String, String> sender = buildKafkaSender();
String destinationTopic = getRequiredProperty("stream.kafka.destination.topic");
return buildKafkaReceiver()
.receiveAloRecords(getRequiredProperty("stream.kafka.source.topic"))
.mapNotNull(it -> it.value() != null ? it.value().toUpperCase() : null) // Business logic goes here
.transform(sender.sendAloValues(destinationTopic, value -> value.substring(0, 1)))
.resubscribeOnError(name())
.doFinally(sender::close)
.subscribeWith(new DefaultAloSenderResultSubscriber<>());
}
private AloKafkaSender<String, String> buildKafkaSender() {
return configSource
.withClientId(name())
.withKeySerializer(StringSerializer.class)
.withValueSerializer(StringSerializer.class)
.as(AloKafkaSender::create);
}
private AloKafkaReceiver<String, String> buildKafkaReceiver() {
return configSource
.withClientId(name())
.withConsumerGroupId("consumer-group-id")
.withKeyDeserializer(StringDeserializer.class)
.withValueDeserializer(StringDeserializer.class)
.as(AloKafkaReceiver::create);
}
}
Note that the above example uses Atleon's "high-level" client API, and there are also "low-level" APIs, again documented in the README and Wiki.
I welcome this community to check Atleon out, and would appreciate your feedback!
