k-raina commented on code in PR #15913:
URL: https://github.com/apache/kafka/pull/15913#discussion_r1629100925
##########
examples/src/main/java/kafka/examples/TransactionalClientDemo.java:
##########
@@ -0,0 +1,153 @@
+package kafka.examples;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+
+import static java.time.Duration.ofSeconds;
+import static java.util.Collections.singleton;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.*;
+import static
org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.*;
+
+/**
+ * This class demonstrates a transactional Kafka client application that
consumes messages from an input topic,
+ * processes them to generate word count statistics, and produces the results
to an output topic.
+ * It utilizes Kafka's transactional capabilities to ensure exactly-once
processing semantics.
+ *
+ * The application continuously polls for records from the input topic,
processes them, and commits the offsets
+ * in a transactional manner. In case of exceptions or errors, it handles them
appropriately, either aborting the
+ * transaction and resetting to the last committed positions, or restarting
the application.
+ *
+ */
+public class TransactionalClientDemo {
+
+ private static final String CONSUMER_GROUP_ID = "my-group-id";
+ private static final String OUTPUT_TOPIC = "output";
+ private static final String INPUT_TOPIC = "input";
+ private static KafkaConsumer<String, String> consumer;
+ private static KafkaProducer<String, String> producer;
+
+ public static void main(String[] args) {
+ initializeApplication();
+
+ boolean isRunning = true;
+ // Continuously poll for records
+ while(isRunning) {
Review Comment:
Updated in commit: 85eb274658b633f3b5de300fd9b2594fe39a3a70
##########
examples/src/main/java/kafka/examples/TransactionalClientDemo.java:
##########
@@ -0,0 +1,153 @@
+package kafka.examples;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+
+import static java.time.Duration.ofSeconds;
+import static java.util.Collections.singleton;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.*;
Review Comment:
Updated in commit: 85eb274658b633f3b5de300fd9b2594fe39a3a70
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
