Atleon is a lightweight reactive stream processing framework that scalably transforms data from any supported infrastructure, and allows sending that data nearly anywhere, while seamlessly maintaining at least once processing guarantees.
Atleon is based on Reactive Streams and backed by Project Reactor. There are two levels of client APIs offered:
- Low-Level: Low-level client APIs are thin reactive wrappers around third-party native infrastructure clients. These APIs are defined in terms of Reactor-native types (e.g.
FluxandMono), and emit elements that contain (or otherwise reference) native infrastructure types. Consumption of messages is facilitated viaReceiverimplementations, with emitted elements containing callbacks for acknowledgement (both positive/successful and negative/failed). Production of messages is facilitated viaSenderimplementations, with emitted elements (implementations ofSenderResult) containing metadata about successes or errors indicating production failure. - High-Level: High-level client APIs decorate low-level clients with an Atleon-native abstraction called
Alo(short for At Least Once).Alofacilitates per-element context which (at minimum) provides access to acknowledgement, and allows for decorating its context with functionality like metrics, distributed tracing, and application-specific metadata. In order to simplify building streams that emit/consumeAlo<T>, high-level clients produce a specialAloFlux<T>type that bridges operations to an underlyingFlux<Alo<T>>, and allows for defining reactive pipelines purely in terms of data typing (T), while providing automatic context propagation.
Atleon documentation and instructions on how to get started are available in the Wiki.
The following is an example of using low-level clients in Atleon:
import io.atleon.kafka.KafkaReceiver;
import io.atleon.kafka.KafkaReceiverOptions;
import io.atleon.kafka.KafkaReceiverRecord;
import io.atleon.kafka.KafkaSender;
import io.atleon.kafka.KafkaSenderOptions;
import io.atleon.kafka.KafkaSenderRecord;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import java.time.Duration;
import java.util.Collections;
public class Example {
public static void main(String[] args) throws Exception {
KafkaSenderOptions<String, String> senderOptions = KafkaSenderOptions.<String, String>newBuilder()
.producerProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
.producerProperty(CommonClientConfigs.CLIENT_ID_CONFIG, "example")
.producerProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName())
.producerProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName())
.build();
KafkaReceiverOptions<String, String> receiverOptions = KafkaReceiverOptions.<String, String>newBuilder()
.consumerProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
.consumerProperty(CommonClientConfigs.CLIENT_ID_CONFIG, "example")
.consumerProperty(ConsumerConfig.GROUP_ID_CONFIG, "consumer-group-id")
.consumerProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
.consumerProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
.build();
// Periodically produce records
KafkaSender<String, String> sender = KafkaSender.create(senderOptions);
Disposable production = Flux.interval(Duration.ofMillis(100))
.map(it -> KafkaSenderRecord.create("source-topic", it.toString(), it.toString(), it))
.transform(sender::send)
.doFinally(__ -> sender.close())
.subscribe();
// Consume produced records
Disposable consumption = KafkaReceiver.create(receiverOptions)
.receiveManual(Collections.singletonList("source-topic"))
.doOnNext(it -> System.out.printf("Consumed record with key=%s and value=%s", it.key(), it.value()))
.subscribe(KafkaReceiverRecord::acknowledge);
System.in.read(); // Added for posterity - Everything above will execute asynchronously
production.dispose(); // Stop producing
consumption.dispose(); // Stop consuming
}
}The next example builds on the availability of low-level clients by switching to high-level Alo clients, removing manual production of data in favor of demonstrating sending/producing transformed messages, and wrapping the resulting stream definition in an implementation of AloStream:
import io.atleon.core.SelfConfigurableAloStream;
import io.atleon.core.DefaultAloSenderResultSubscriber;
import io.atleon.kafka.AloKafkaReceiver;
import io.atleon.kafka.AloKafkaSender;
import io.atleon.kafka.KafkaConfigSource;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import reactor.core.Disposable;
public class MyStream extends SelfConfigurableAloStream {
private final KafkaConfigSource configSource;
private final String sourceTopic;
private final String destinationTopic;
public MyStream(KafkaConfigSource configSource, String sourceTopic, String destinationTopic) {
this.configSource = configSource;
this.sourceTopic = sourceTopic;
this.destinationTopic = destinationTopic;
}
@Override
public Disposable startDisposable() {
AloKafkaSender<String, String> sender = buildKafkaSender();
return buildKafkaReceiver()
.receiveAloRecords(sourceTopic)
.mapNotNull(it -> it.value() != null ? it.value().toUpperCase() : null) // Business logic goes here
.transform(sender.sendAloValues(destinationTopic, message -> message.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(StringSerializer.class)
.withValueDeserializer(StringSerializer.class)
.as(AloKafkaReceiver::create);
}
}Atleon has built-in integration with Spring, where a fully configured AloStream looks like the following:
pom.xml:
<dependencies>
<dependency>
<groupId>io.atleon</groupId>
<artifactId>atleon-kafka</artifactId> <!-- Include infrastructure client(s) -->
<version>${atleon.version}</version>
</dependency>
<dependency>
<groupId>io.atleon</groupId>
<artifactId>atleon-spring</artifactId> <!-- Application binding(s) -->
<version>${atleon.version}</version>
</dependency>
</dependencies>application.yml:
atleon:
config.sources:
- name: kafkaConfigSource
type: kafka
bootstrap.servers: localhost:9092
stream:
kafka:
destination.topic: output
source.topic: inputMyStream.java:
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);
}
@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, message -> message.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);
}
}The examples module contains runnable classes showing Atleon in action and intended usage.
Atleon is built using Maven. Installing Maven locally is optional as you can use the Maven Wrapper:
./mvnw clean verify
Atleon makes use of Testcontainers for some unit tests. Testcontainers is based on Docker, so successfully building Atleon requires Docker to be running locally.
Please refer to CONTRIBUTING for information on how to contribute to Atleon
This project is available under the Apache 2.0 License.