Page History

  Set<String> topics = buildSubscription(); 	
  KafkaConsumer consumer = new KafkaConsumer<>(consumerConfig);
  KafkaProducer producer = new KafkaProducer();
 
 ConsumerGroupMetadata groupMetadata // Will access consumer internal state. Only called once in the app's life cycle.
  // Note that this is a blocking call until consumer successfully joins the group.= null;

  consumer.subscribe(topics, new ConsumerRebalanceListener() {
    void onGroupRejoined(ConsumerGroupMetadata updatedMetadata) {
      consumerGroupMetadata = updatedMetadata;
    }
  });

  producer.initTransactions(consumer);	
  while (true) {
    // Read some records from the consumer and collect the offsets to commit
    ConsumerRecords consumed = consumer.poll(Duration.ofMillis(5000)); // This will be the fencing point if there are pending offsets for the first time.
    Map<TopicPartition, OffsetAndMetadata> consumedOffsets = offsets(consumed);

    // Do some processing and build the records we want to produce
    List<ProducerRecord> processed = process(consumed);

    // Write the records and commit offsets under a single transaction
    producer.beginTransaction();
    for (ProducerRecord record : processed)
      producer.send(record);
    )
      producer.send(record);
    
	if (groupMetadata == null) {
		throw new IllegalStateException("Consumer poll should be blocked until successfully joined the group and get updated metadata");
	}
    producer.sendOffsetsToTransaction(consumedOffsets, groupMetadata);

    producer.commitTransaction();
  }