Using Kafka Streams

// org.apache.kafka:kafka-streams:3.9.0
import org.apache.kafka.streams.*;

StreamsBuilder b = new StreamsBuilder();
b.stream("input").to("output");
KafkaStreams ks = new KafkaStreams(b.build(), props);
ks.start();

Topology t = builder.build();
System.out.println(t.describe());

stream.filter((k, v) -> v.getAmount() > 100)
      .mapValues(v -> v.getAmount() * 1.1);

KTable<String, Long> counts = stream
    .groupByKey()
    .count();

orders.join(customers,
    (order, cust) -> enrich(order, cust));

stream.groupByKey()
  .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
  .count();

ReadOnlyKeyValueStore<String, Long> store =
  ks.store(StoreQueryParameters.fromNameAndType(
    "counts", QueryableStoreTypes.keyValueStore()));
Long n = store.get("key1");

stream.peek((k, v) -> log.info("{}={}", k, v))
      .to("output");

builder.stream("in").process(() -> new Processor<>() {
  public void process(Record<String,String> r) {
    context.forward(r.withValue(r.value().toUpperCase()));
  }
});

ks.start();
Runtime.getRuntime().addShutdownHook(new Thread(ks::close));

ks.setUncaughtExceptionHandler(e ->
  StreamThreadExceptionResponse.REPLACE_THREAD);