Messages from and to Apache Kafka

libraryDependencies += "org.apache.pekko" %% "pekko-projection-kafka" % "1.1.0"

copysourceimport org.apache.pekko.kafka.ConsumerSettings;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

String bootstrapServers = "localhost:9092";
String groupId = "group-wordcount";
String topicName = "words";
ConsumerSettings<String, String> consumerSettings =
    ConsumerSettings.create(system, new StringDeserializer(), new StringDeserializer())
        .withBootstrapServers(bootstrapServers)
        .withGroupId(groupId)
        .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

SourceProvider<MergeableOffset<Long>, ConsumerRecord<String, String>> sourceProvider =
    KafkaSourceProvider.create(system, consumerSettings, Collections.singleton(topicName));

copysourcefinal HibernateSessionFactory sessionProvider = new HibernateSessionFactory();

ProjectionId projectionId = ProjectionId.of("WordCount", "wordcount-1");
ExactlyOnceProjection<MergeableOffset<Long>, ConsumerRecord<String, String>> projection =
    JdbcProjection.exactlyOnce(
        projectionId,
        sourceProvider,
        sessionProvider::newInstance,
        () -> new WordCountJdbcHandler(wordRepository),
        system);

copysourcepublic class WordCountJdbcHandler
    extends JdbcHandler<ConsumerRecord<String, String>, HibernateJdbcSession> {
  private Logger logger = LoggerFactory.getLogger(getClass());
  private WordRepository wordRepository;

  public WordCountJdbcHandler(WordRepository wordRepository) {
    this.wordRepository = wordRepository;
  }

  @Override
  public void process(HibernateJdbcSession session, ConsumerRecord<String, String> envelope)
      throws Exception {
    String word = envelope.value();
    wordRepository.increment(session.entityManager, word);
  }
}

copysourceinterface WordRepository {
  void increment(EntityManager entityManager, String word);
}

copysourceclass WordPublisher extends Handler<WordEnvelope> {
  private final Logger logger = LoggerFactory.getLogger(getClass());
  private final String topic;
  private final SendProducer<String, String> sendProducer;

  public WordPublisher(String topic, SendProducer<String, String> sendProducer) {
    this.topic = topic;
    this.sendProducer = sendProducer;
  }

  @Override
  public CompletionStage<Done> process(WordEnvelope envelope) {
    String word = envelope.word;
    // using the word as the key and `DefaultPartitioner` will select partition based on the key
    // so that same word always ends up in same partition
    String key = word;
    ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, key, word);
    CompletionStage<RecordMetadata> result = sendProducer.send(producerRecord);
    CompletionStage<Done> done =
        result.thenApply(
            recordMetadata -> {
              logger.info(
                  "Published word [{}] to topic/partition {}/{}",
                  word,
                  topic,
                  recordMetadata.partition());
              return Done.getInstance();
            });
    return done;
  }
}

copysourceimport org.apache.kafka.common.serialization.StringSerializer;
import org.apache.pekko.kafka.ProducerSettings;
import org.apache.pekko.kafka.javadsl.SendProducer;

String bootstrapServers = "localhost:9092";
String topicName = "words";
ProducerSettings<String, String> producerSettings =
    ProducerSettings.create(system, new StringSerializer(), new StringSerializer())
        .withBootstrapServers(bootstrapServers);
// FIXME classicSystem might not be needed in later Pekko Connectors Kafka version?
SendProducer<String, String> sendProducer =
    new SendProducer<>(producerSettings, system.classicSystem());

copysourceWordSource sourceProvider = new WordSource();
HibernateSessionFactory sessionProvider = new HibernateSessionFactory();

ProjectionId projectionId = ProjectionId.of("PublishWords", "words");
Projection<WordEnvelope> projection =
    JdbcProjection.atLeastOnceAsync(
        projectionId,
        sourceProvider,
        sessionProvider::newInstance,
        () -> new WordPublisher(topicName, sendProducer),
        system);

copysourcepublic class WordEnvelope {
  public final Long offset;
  public final String word;

  public WordEnvelope(Long offset, String word) {
    this.offset = offset;
    this.word = word;
  }
}

class WordSource extends SourceProvider<Long, WordEnvelope> {

  private final Source<WordEnvelope, NotUsed> src =
      Source.from(
          Arrays.asList(
              new WordEnvelope(1L, "abc"),
              new WordEnvelope(2L, "def"),
              new WordEnvelope(3L, "ghi"),
              new WordEnvelope(4L, "abc")));

  @Override
  public CompletionStage<Source<WordEnvelope, NotUsed>> source(
      Supplier<CompletionStage<Optional<Long>>> offset) {
    return offset
        .get()
        .thenApply(
            o -> {
              if (o.isPresent())
                return src.dropWhile(envelope -> envelope.offset <= o.get())
                    .throttle(1, Duration.ofSeconds(1));
              else return src.throttle(1, Duration.ofSeconds(1));
            });
  }

  @Override
  public Long extractOffset(WordEnvelope envelope) {
    return envelope.offset;
  }

  @Override
  public long extractCreationTime(WordEnvelope envelope) {
    return 0L;
  }
}

copysourceimport org.apache.kafka.common.serialization.StringSerializer;
import org.apache.pekko.kafka.ProducerSettings;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.pekko.kafka.ProducerMessage;
import org.apache.pekko.kafka.javadsl.Producer;
import org.apache.pekko.stream.javadsl.FlowWithContext;
import org.apache.pekko.projection.ProjectionContext;

String bootstrapServers = "localhost:9092";
String topicName = "words";

ProducerSettings<String, String> producerSettings =
    ProducerSettings.create(system, new StringSerializer(), new StringSerializer())
        .withBootstrapServers(bootstrapServers);

FlowWithContext<WordEnvelope, ProjectionContext, Done, ProjectionContext, NotUsed>
    producerFlow =
        FlowWithContext.<WordEnvelope, ProjectionContext>create()
            .map(
                wordEnv ->
                    ProducerMessage.single(
                        new ProducerRecord<String, String>(
                            topicName, wordEnv.word, wordEnv.word)))
            .via(Producer.flowWithContext(producerSettings))
            .map(__ -> Done.getInstance());

copysourceWordSource sourceProvider = new WordSource();
HibernateSessionFactory sessionProvider = new HibernateSessionFactory();

ProjectionId projectionId = ProjectionId.of("PublishWords", "words");
Projection<WordEnvelope> projection =
    JdbcProjection.atLeastOnceFlow(
        projectionId, sourceProvider, sessionProvider::newInstance, producerFlow, system);