You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sh...@apache.org on 2023/05/02 12:28:50 UTC
[kafka] branch trunk updated: KAFKA-14752: Kafka examples improvements - consumer changes (#13514)
This is an automated email from the ASF dual-hosted git repository.
showuon pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 33012b5ec34 KAFKA-14752: Kafka examples improvements - consumer changes (#13514)
33012b5ec34 is described below
commit 33012b5ec34305a5133eb6e9e2fb6e8c3178f3b3
Author: Federico Valeri <fe...@gmail.com>
AuthorDate: Tue May 2 14:28:42 2023 +0200
KAFKA-14752: Kafka examples improvements - consumer changes (#13514)
KAFKA-14752: Kafka examples improvements - consumer changes
This is extracted from the original PR for better review.
https://github.com/apache/kafka/pull/13492
Signed-off-by: Federico Valeri <fe...@gmail.com>
Reviewers: Christo Lolov <ch...@gmail.com>, Luke Chen <sh...@gmail.com>
---
.../src/main/java/kafka/examples/Consumer.java | 164 +++++++++++++--------
.../examples/ExactlyOnceMessageProcessor.java | 6 +-
.../kafka/examples/KafkaConsumerProducerDemo.java | 3 +-
.../java/kafka/examples/KafkaExactlyOnceDemo.java | 3 +-
examples/src/main/java/kafka/examples/Utils.java | 106 +++++++++++++
5 files changed, 220 insertions(+), 62 deletions(-)
diff --git a/examples/src/main/java/kafka/examples/Consumer.java b/examples/src/main/java/kafka/examples/Consumer.java
index 7a5dc05b7a7..abb69134e72 100644
--- a/examples/src/main/java/kafka/examples/Consumer.java
+++ b/examples/src/main/java/kafka/examples/Consumer.java
@@ -21,97 +21,145 @@ import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
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.NoOffsetForPartitionException;
+import org.apache.kafka.clients.consumer.OffsetOutOfRangeException;
+import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.errors.AuthorizationException;
+import org.apache.kafka.common.errors.RecordDeserializationException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collection;
-import java.util.Collections;
import java.util.Optional;
import java.util.Properties;
+import java.util.UUID;
import java.util.concurrent.CountDownLatch;
+import static java.util.Collections.singleton;
+
/**
- * A simple consumer thread that demonstrate subscribe and poll use case. The thread subscribes to a topic,
- * then runs a loop to poll new messages, and print the message out. The thread closes until the target {@code
- * numMessageToConsume} is hit or catching an exception.
+ * A simple consumer thread that subscribes to a topic, fetches new records and prints them.
+ * The thread does not stop until all records are completed or an exception is raised.
*/
public class Consumer extends Thread implements ConsumerRebalanceListener {
- private final KafkaConsumer<Integer, String> consumer;
+ private final String bootstrapServers;
private final String topic;
private final String groupId;
- private final int numMessageToConsume;
- private int messageRemaining;
+ private final Optional<String> instanceId;
+ private final boolean readCommitted;
+ private final int numRecords;
private final CountDownLatch latch;
+ private volatile boolean closed;
+ private int remainingRecords;
- public Consumer(final String topic,
- final String groupId,
- final Optional<String> instanceId,
- final boolean readCommitted,
- final int numMessageToConsume,
- final CountDownLatch latch) {
- super("KafkaConsumerExample");
- this.groupId = groupId;
- Properties props = new Properties();
- props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT);
- props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
- instanceId.ifPresent(id -> props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, id));
- props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
- props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
- if (readCommitted) {
- props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
- }
- props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-
- consumer = new KafkaConsumer<>(props);
+ public Consumer(String threadName,
+ String bootstrapServers,
+ String topic,
+ String groupId,
+ Optional<String> instanceId,
+ boolean readCommitted,
+ int numRecords,
+ CountDownLatch latch) {
+ super(threadName);
+ this.bootstrapServers = bootstrapServers;
this.topic = topic;
- this.numMessageToConsume = numMessageToConsume;
- this.messageRemaining = numMessageToConsume;
+ this.groupId = groupId;
+ this.instanceId = instanceId;
+ this.readCommitted = readCommitted;
+ this.numRecords = numRecords;
+ this.remainingRecords = numRecords;
this.latch = latch;
}
- KafkaConsumer<Integer, String> get() {
- return consumer;
- }
-
@Override
public void run() {
- try {
- System.out.println("Subscribe to:" + this.topic);
- consumer.subscribe(Collections.singletonList(this.topic), this);
- do {
- doWork();
- } while (messageRemaining > 0);
- System.out.println(groupId + " finished reading " + numMessageToConsume + " messages");
- } catch (WakeupException e) {
- // swallow the wakeup
- } catch (Exception e) {
- System.out.println("Unexpected termination, exception thrown:" + e);
- } finally {
- shutdown();
+ // the consumer instance is NOT thread safe
+ try (KafkaConsumer<Integer, String> consumer = createKafkaConsumer()) {
+ // subscribes to a list of topics to get dynamically assigned partitions
+ // this class implements the rebalance listener that we pass here to be notified of such events
+ consumer.subscribe(singleton(topic), this);
+ Utils.printOut("Subscribed to %s", topic);
+ while (!closed && remainingRecords > 0) {
+ try {
+ // if required, poll updates partition assignment and invokes the configured rebalance listener
+ // then tries to fetch records sequentially using the last committed offset or auto.offset.reset policy
+ // returns immediately if there are records or times out returning an empty record set
+ // the next poll must be called within session.timeout.ms to avoid group rebalance
+ ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(1));
+ for (ConsumerRecord<Integer, String> record : records) {
+ Utils.maybePrintRecord(numRecords, record);
+ }
+ remainingRecords -= records.count();
+ } catch (AuthorizationException | UnsupportedVersionException
+ | RecordDeserializationException e) {
+ // we can't recover from these exceptions
+ Utils.printErr(e.getMessage());
+ shutdown();
+ } catch (OffsetOutOfRangeException | NoOffsetForPartitionException e) {
+ // invalid or no offset found without auto.reset.policy
+ Utils.printOut("Invalid or no offset found, using latest");
+ consumer.seekToEnd(e.partitions());
+ consumer.commitSync();
+ } catch (KafkaException e) {
+ // log the exception and try to continue
+ Utils.printErr(e.getMessage());
+ }
+ }
+ } catch (Throwable e) {
+ Utils.printOut("Unhandled exception");
+ e.printStackTrace();
}
+ Utils.printOut("Fetched %d records", numRecords - remainingRecords);
+ shutdown();
}
- public void doWork() {
- ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(1));
- for (ConsumerRecord<Integer, String> record : records) {
- System.out.println(groupId + " received message : from partition " + record.partition() + ", (" + record.key() + ", " + record.value() + ") at offset " + record.offset());
+
+ public void shutdown() {
+ if (!closed) {
+ closed = true;
+ latch.countDown();
}
- messageRemaining -= records.count();
}
- public void shutdown() {
- this.consumer.close();
- latch.countDown();
+ public KafkaConsumer<Integer, String> createKafkaConsumer() {
+ Properties props = new Properties();
+ // bootstrap server config is required for consumer to connect to brokers
+ props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ // client id is not required, but it's good to track the source of requests beyond just ip/port
+ // by allowing a logical application name to be included in server-side request logging
+ props.put(ConsumerConfig.CLIENT_ID_CONFIG, "client-" + UUID.randomUUID());
+ // consumer group id is required when we use subscribe(topics) for group management
+ props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ // sets static membership to improve availability (e.g. rolling restart)
+ instanceId.ifPresent(id -> props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, id));
+ // disables auto commit when EOS is enabled, because offsets are committed with the transaction
+ props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, readCommitted ? "false" : "true");
+ // key and value are just byte arrays, so we need to set appropriate deserializers
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ if (readCommitted) {
+ // skips ongoing and aborted transactions
+ props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
+ }
+ // sets the reset offset policy in case of invalid or no offset
+ props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ return new KafkaConsumer<>(props);
}
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
- System.out.println("Revoking partitions:" + partitions);
+ Utils.printOut("Revoked partitions: %s", partitions);
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
- System.out.println("Assigning partitions:" + partitions);
+ Utils.printOut("Assigned partitions: %s", partitions);
+ }
+
+ @Override
+ public void onPartitionsLost(Collection<TopicPartition> partitions) {
+ Utils.printOut("Lost partitions: %s", partitions);
}
}
diff --git a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java
index 8f31b196267..6de6ab99d26 100644
--- a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java
+++ b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java
@@ -70,8 +70,10 @@ public class ExactlyOnceMessageProcessor extends Thread {
// Consumer must be in read_committed mode, which means it won't be able to read uncommitted data.
// Consumer could optionally configure groupInstanceId to avoid unnecessary rebalances.
this.groupInstanceId = "Txn-consumer-" + instanceIdx;
- consumer = new Consumer(inputTopic, "Eos-consumer",
- Optional.of(groupInstanceId), READ_COMMITTED, -1, null).get();
+ boolean readCommitted = true;
+ consumer = new Consumer(
+ "processor-consumer", KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT, inputTopic, "processor-group", Optional.of(groupInstanceId), readCommitted, -1, null)
+ .createKafkaConsumer();
this.latch = latch;
}
diff --git a/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java b/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java
index 9fc911acac0..4087367f899 100644
--- a/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java
+++ b/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java
@@ -29,7 +29,8 @@ public class KafkaConsumerProducerDemo {
Producer producerThread = new Producer(KafkaProperties.TOPIC, isAsync, null, false, 10000, -1, latch);
producerThread.start();
- Consumer consumerThread = new Consumer(KafkaProperties.TOPIC, "DemoConsumer", Optional.empty(), false, 10000, latch);
+ Consumer consumerThread = new Consumer(
+ "consumer", KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT, KafkaProperties.TOPIC, "DemoConsumer", Optional.empty(), false, 10000, latch);
consumerThread.start();
if (!latch.await(5, TimeUnit.MINUTES)) {
diff --git a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java
index 8fbab1e6226..03ee8e25498 100644
--- a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java
+++ b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java
@@ -114,7 +114,8 @@ public class KafkaExactlyOnceDemo {
CountDownLatch consumeLatch = new CountDownLatch(1);
/* Stage 4: consume all processed messages to verify exactly once */
- Consumer consumerThread = new Consumer(OUTPUT_TOPIC, "Verify-consumer", Optional.empty(), true, numRecords, consumeLatch);
+ Consumer consumerThread = new Consumer(
+ "consumer", "DemoConsumer", OUTPUT_TOPIC, "Verify-consumer", Optional.empty(), true, numRecords, consumeLatch);
consumerThread.start();
if (!consumeLatch.await(5, TimeUnit.MINUTES)) {
diff --git a/examples/src/main/java/kafka/examples/Utils.java b/examples/src/main/java/kafka/examples/Utils.java
new file mode 100644
index 00000000000..8846879f67d
--- /dev/null
+++ b/examples/src/main/java/kafka/examples/Utils.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.examples;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.errors.TopicExistsException;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static java.lang.String.format;
+
+public class Utils {
+ private Utils() {
+ }
+
+ public static void printHelp(String message, Object... args) {
+ System.out.println(format(message, args));
+ }
+
+ public static void printOut(String message, Object... args) {
+ System.out.printf("%s - %s%n", Thread.currentThread().getName(), format(message, args));
+ }
+
+ public static void printErr(String message, Object... args) {
+ System.err.printf("%s - %s%n", Thread.currentThread().getName(), format(message, args));
+ }
+
+ public static void maybePrintRecord(long numRecords, ConsumerRecord<Integer, String> record) {
+ maybePrintRecord(numRecords, record.key(), record.value(), record.topic(), record.partition(), record.offset());
+ }
+
+ public static void maybePrintRecord(long numRecords, int key, String value, RecordMetadata metadata) {
+ maybePrintRecord(numRecords, key, value, metadata.topic(), metadata.partition(), metadata.offset());
+ }
+
+ private static void maybePrintRecord(long numRecords, int key, String value, String topic, int partition, long offset) {
+ // we only print 10 records when there are 20 or more to send
+ if (key % Math.max(1, numRecords / 10) == 0) {
+ printOut("Sample: record(%d, %s), partition(%s-%d), offset(%d)", key, value, topic, partition, offset);
+ }
+ }
+
+ public static void recreateTopics(String bootstrapServers, int numPartitions, String... topicNames) {
+ Properties props = new Properties();
+ props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ props.put(AdminClientConfig.CLIENT_ID_CONFIG, "client-" + UUID.randomUUID());
+ try (Admin admin = Admin.create(props)) {
+ // delete topics if present
+ try {
+ admin.deleteTopics(Arrays.asList(topicNames)).all().get();
+ } catch (ExecutionException e) {
+ if (!(e.getCause() instanceof UnknownTopicOrPartitionException)) {
+ throw e;
+ }
+ printErr("Topics deletion error: %s", e.getCause());
+ }
+ printOut("Deleted topics: %s", Arrays.toString(topicNames));
+ // create topics in a retry loop
+ while (true) {
+ // use default RF to avoid NOT_ENOUGH_REPLICAS error with minISR > 1
+ short replicationFactor = -1;
+ List<NewTopic> newTopics = Arrays.stream(topicNames)
+ .map(name -> new NewTopic(name, numPartitions, replicationFactor))
+ .collect(Collectors.toList());
+ try {
+ admin.createTopics(newTopics).all().get();
+ printOut("Created topics: %s", Arrays.toString(topicNames));
+ break;
+ } catch (ExecutionException e) {
+ if (!(e.getCause() instanceof TopicExistsException)) {
+ throw e;
+ }
+ printOut("Waiting for topics metadata cleanup");
+ TimeUnit.MILLISECONDS.sleep(1_000);
+ }
+ }
+ } catch (Throwable e) {
+ throw new RuntimeException("Topics creation error", e);
+ }
+ }
+}