From b4a6a3a72887c18bed1e3cafbef5b68fd8d47ca5 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Wed, 22 Jan 2020 16:46:08 -0800 Subject: [PATCH 01/11] add eos example --- .../controller/ControllerEventManager.scala | 2 +- .../src/main/scala/kafka/log/LogCleaner.scala | 2 +- .../kafka/utils/ShutdownableThread.scala | 4 +- .../DynamicBrokerReconfigurationTest.scala | 4 +- .../kafka/TestPurgatoryPerformance.scala | 2 +- examples/README | 1 - .../main/java/kafka/examples/Consumer.java | 12 +++- .../examples/KafkaConsumerProducerDemo.java | 8 +-- .../kafka/examples/KafkaExactlyOnceDemo.java | 68 +++++++++++++++++++ .../main/java/kafka/examples/Producer.java | 9 ++- 10 files changed, 97 insertions(+), 15 deletions(-) create mode 100644 examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java diff --git a/core/src/main/scala/kafka/controller/ControllerEventManager.scala b/core/src/main/scala/kafka/controller/ControllerEventManager.scala index b04b85f1fa1c8..eccb4a113e038 100644 --- a/core/src/main/scala/kafka/controller/ControllerEventManager.scala +++ b/core/src/main/scala/kafka/controller/ControllerEventManager.scala @@ -112,7 +112,7 @@ class ControllerEventManager(controllerId: Int, def isEmpty: Boolean = queue.isEmpty - class ControllerEventThread(name: String) extends ShutdownableThread(name = name, isInterruptible = false) { + class ControllerEventThread(name: String) extends ShutdownableThread(name = name, isInterruptable = false) { logIdent = s"[ControllerEventThread controllerId=$controllerId] " override def doWork(): Unit = { diff --git a/core/src/main/scala/kafka/log/LogCleaner.scala b/core/src/main/scala/kafka/log/LogCleaner.scala index 312483dfeefca..46c0ef876b16c 100644 --- a/core/src/main/scala/kafka/log/LogCleaner.scala +++ b/core/src/main/scala/kafka/log/LogCleaner.scala @@ -269,7 +269,7 @@ class LogCleaner(initialConfig: CleanerConfig, * choosing the dirtiest log, cleaning it, and then swapping in the cleaned segments. */ private[log] class CleanerThread(threadId: Int) - extends ShutdownableThread(name = s"kafka-log-cleaner-thread-$threadId", isInterruptible = false) { + extends ShutdownableThread(name = s"kafka-log-cleaner-thread-$threadId", isInterruptable = false) { protected override def loggerName = classOf[LogCleaner].getName diff --git a/core/src/main/scala/kafka/utils/ShutdownableThread.scala b/core/src/main/scala/kafka/utils/ShutdownableThread.scala index 0ca21c4ab166d..c16e4953d5209 100644 --- a/core/src/main/scala/kafka/utils/ShutdownableThread.scala +++ b/core/src/main/scala/kafka/utils/ShutdownableThread.scala @@ -21,7 +21,7 @@ import java.util.concurrent.{CountDownLatch, TimeUnit} import org.apache.kafka.common.internals.FatalExitError -abstract class ShutdownableThread(val name: String, val isInterruptible: Boolean = true) +abstract class ShutdownableThread(val name: String, val isInterruptable: Boolean = true) extends Thread(name) with Logging { this.setDaemon(false) this.logIdent = "[" + name + "]: " @@ -50,7 +50,7 @@ abstract class ShutdownableThread(val name: String, val isInterruptible: Boolean if (isRunning) { info("Shutting down") shutdownInitiated.countDown() - if (isInterruptible) + if (isInterruptable) interrupt() true } else diff --git a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala index 3e7fa15a15d50..36d13e72fecec 100644 --- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala +++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala @@ -1613,7 +1613,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet } private class ProducerThread(clientId: String, retries: Int) - extends ShutdownableThread(clientId, isInterruptible = false) { + extends ShutdownableThread(clientId, isInterruptable = false) { private val producer = ProducerBuilder().maxRetries(retries).clientId(clientId).build() val lastSent = new ConcurrentHashMap[Int, Int]() @@ -1634,7 +1634,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet } } - private class ConsumerThread(producerThread: ProducerThread) extends ShutdownableThread("test-consumer", isInterruptible = false) { + private class ConsumerThread(producerThread: ProducerThread) extends ShutdownableThread("test-consumer", isInterruptable = false) { private val consumer = ConsumerBuilder("group1").enableAutoCommit(true).build() val lastReceived = new ConcurrentHashMap[Int, Int]() val missingRecords = new ConcurrentLinkedQueue[Int]() diff --git a/core/src/test/scala/other/kafka/TestPurgatoryPerformance.scala b/core/src/test/scala/other/kafka/TestPurgatoryPerformance.scala index 4b540a4f1a829..0ad5619f850dd 100644 --- a/core/src/test/scala/other/kafka/TestPurgatoryPerformance.scala +++ b/core/src/test/scala/other/kafka/TestPurgatoryPerformance.scala @@ -256,7 +256,7 @@ object TestPurgatoryPerformance { private class CompletionQueue { private[this] val delayQueue = new DelayQueue[Scheduled]() - private[this] val thread = new ShutdownableThread(name = "completion thread", isInterruptible = false) { + private[this] val thread = new ShutdownableThread(name = "completion thread", isInterruptable = false) { override def doWork(): Unit = { val scheduled = delayQueue.poll(100, TimeUnit.MILLISECONDS) if (scheduled != null) { diff --git a/examples/README b/examples/README index f6e3410a2a0aa..887cbd6b76f72 100644 --- a/examples/README +++ b/examples/README @@ -6,4 +6,3 @@ To run the demo: 2. For simple consumer demo, `run bin/java-simple-consumer-demo.sh` 3. For unlimited sync-producer-consumer run, `run bin/java-producer-consumer-demo.sh sync` 4. For unlimited async-producer-consumer run, `run bin/java-producer-consumer-demo.sh` - diff --git a/examples/src/main/java/kafka/examples/Consumer.java b/examples/src/main/java/kafka/examples/Consumer.java index 26d6e23a3f8e8..6556d27008b92 100644 --- a/examples/src/main/java/kafka/examples/Consumer.java +++ b/examples/src/main/java/kafka/examples/Consumer.java @@ -21,6 +21,7 @@ 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.common.IsolationLevel; import java.time.Duration; import java.util.Collections; @@ -30,7 +31,7 @@ public class Consumer extends ShutdownableThread { private final KafkaConsumer consumer; private final String topic; - public Consumer(String topic) { + public Consumer(String topic, boolean readCommitted) { super("KafkaConsumerExample", false); Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT); @@ -40,11 +41,18 @@ public Consumer(String topic) { props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000"); 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"); + } consumer = new KafkaConsumer<>(props); this.topic = topic; } + KafkaConsumer get() { + return consumer; + } + @Override public void doWork() { consumer.subscribe(Collections.singletonList(this.topic)); @@ -60,7 +68,7 @@ public String name() { } @Override - public boolean isInterruptible() { + public boolean isInterruptable() { return false; } } diff --git a/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java b/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java index f42ed6f4abd1e..ba68fd16c4fdd 100644 --- a/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java +++ b/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java @@ -19,11 +19,11 @@ public class KafkaConsumerProducerDemo { public static void main(String[] args) { boolean isAsync = args.length == 0 || !args[0].trim().equalsIgnoreCase("sync"); - Producer producerThread = new Producer(KafkaProperties.TOPIC, isAsync); - producerThread.start(); - + Producer producerThread = new Producer(KafkaProperties.TOPIC, isAsync, null); + producerThread.start(); + Consumer consumerThread = new Consumer(KafkaProperties.TOPIC); - consumerThread.start(); + consumerThread.start(); } } diff --git a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java new file mode 100644 index 0000000000000..3ca72e3f46ea3 --- /dev/null +++ b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java @@ -0,0 +1,68 @@ +package kafka.examples; + +import org.apache.kafka.clients.consumer.CommitFailedException; +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 org.apache.kafka.common.errors.FencedInstanceIdException; +import org.apache.kafka.common.errors.ProducerFencedException; + +import java.time.Duration; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; + +/** + * A demo class for how to write a customized EOS app. It takes a consume-process-produce loop + * The things to pay attention to beyond a general consumer + producer app are: + * 1. Define a unique transactional.id for your app + * 2. Turn on read_committed isolation level + */ +public class KafkaExactlyOnceDemo { + + private static final String OUTPUT_TOPIC = "output-topic"; + private static final String TRANSACTIONAL_ID = "transactional-id-" + UUID.randomUUID(); + private static final boolean READ_COMMITTED = true; + + public static void main(String[] args) { + KafkaProducer producer = new Producer(KafkaProperties.TOPIC, true, TRANSACTIONAL_ID).get(); + producer.initTransactions(); + + KafkaConsumer consumer = new Consumer(KafkaProperties.TOPIC, READ_COMMITTED).get(); + consumer.subscribe(Collections.singleton(KafkaProperties.TOPIC)); + while (true) { + ConsumerRecords records = consumer.poll(Duration.ofMillis(200)); + if (records.count() > 0) { + try { + producer.beginTransaction(); + for (ConsumerRecord record : records) { + ConsumerRecord customizedRecord = transform(record); + producer.send(new ProducerRecord<>(OUTPUT_TOPIC, customizedRecord.key(), customizedRecord.value())); + } + + Map positions = new HashMap<>(); + for (TopicPartition topicPartition : consumer.assignment()) { + positions.put(topicPartition, new OffsetAndMetadata(consumer.position(topicPartition), null)); + } + producer.sendOffsetsToTransaction(positions, consumer.groupMetadata()); + producer.commitTransaction(); + } catch (CommitFailedException e) { + producer.abortTransaction(); + } catch (ProducerFencedException | FencedInstanceIdException e) { + throw new KafkaException("Encountered fatal error during processing: " + e.getMessage()); + } + } + } + } + + private static ConsumerRecord transform(ConsumerRecord record) { + // Customized business logic here + return record; + } +} diff --git a/examples/src/main/java/kafka/examples/Producer.java b/examples/src/main/java/kafka/examples/Producer.java index b6998c58ac756..e37f4cd9fe160 100644 --- a/examples/src/main/java/kafka/examples/Producer.java +++ b/examples/src/main/java/kafka/examples/Producer.java @@ -32,17 +32,24 @@ public class Producer extends Thread { private final String topic; private final Boolean isAsync; - public Producer(String topic, Boolean isAsync) { + public Producer(String topic, Boolean isAsync, String transactionalId) { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT); props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoProducer"); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + if (transactionalId != null) { + props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId); + } producer = new KafkaProducer<>(props); this.topic = topic; this.isAsync = isAsync; } + KafkaProducer get() { + return producer; + } + public void run() { int messageNo = 1; while (true) { From 0cf5868943fe5f4cde514d14b049d4b7a13a32b5 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Fri, 31 Jan 2020 16:39:52 -0500 Subject: [PATCH 02/11] extend producer demo with even number keys --- .../kafka/examples/KafkaExactlyOnceDemo.java | 61 ++++++++++++++++--- .../main/java/kafka/examples/Producer.java | 4 +- 2 files changed, 56 insertions(+), 9 deletions(-) diff --git a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java index 3ca72e3f46ea3..d60e05ba086f7 100644 --- a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java +++ b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java @@ -1,6 +1,7 @@ package kafka.examples; import org.apache.kafka.clients.consumer.CommitFailedException; +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; @@ -13,11 +14,14 @@ import org.apache.kafka.common.errors.ProducerFencedException; import java.time.Duration; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.UUID; +import static java.util.Collections.singleton; + /** * A demo class for how to write a customized EOS app. It takes a consume-process-produce loop * The things to pay attention to beyond a general consumer + producer app are: @@ -31,26 +35,56 @@ public class KafkaExactlyOnceDemo { private static final boolean READ_COMMITTED = true; public static void main(String[] args) { + if (args.length < 1) { + throw new IllegalArgumentException("Should accept at least one parameter"); + } else if (args[0].equals("standaloneMode") && args.length != 2) { + throw new IllegalArgumentException("Should have specified partition in standalone mode"); + } + KafkaProducer producer = new Producer(KafkaProperties.TOPIC, true, TRANSACTIONAL_ID).get(); + // Init transactions call should always happen first in order to clear zombie transactions. producer.initTransactions(); KafkaConsumer consumer = new Consumer(KafkaProperties.TOPIC, READ_COMMITTED).get(); - consumer.subscribe(Collections.singleton(KafkaProperties.TOPIC)); - while (true) { + // Under group mode, topic based subscription is sufficient as Consumers are safe to work transactionally after 2.5. + // Under standalone mode, user needs to manually assign the topic partitions and make sure the assignment is unique + // across the consumer group. + if (args[0].equals("groupMode")) { + consumer.subscribe(Collections.singleton(KafkaProperties.TOPIC), new ConsumerRebalanceListener() { + @Override + public void onPartitionsRevoked(Collection partitions) { + + } + + @Override + public void onPartitionsAssigned(Collection partitions) { + System.out.println("Get partition assignment " + partitions); + } + }); + } else { + consumer.assign(Collections.singletonList(new TopicPartition(KafkaProperties.TOPIC, + Integer.valueOf(args[1])))); + } + long messageRemaining = messagesRemaining(consumer); + + while (messageRemaining > 0) { ConsumerRecords records = consumer.poll(Duration.ofMillis(200)); if (records.count() > 0) { try { + // Begin a new transaction session. producer.beginTransaction(); for (ConsumerRecord record : records) { - ConsumerRecord customizedRecord = transform(record); - producer.send(new ProducerRecord<>(OUTPUT_TOPIC, customizedRecord.key(), customizedRecord.value())); + ProducerRecord customizedRecord = transform(record); + // Send records to the downstream. + producer.send(customizedRecord); } - Map positions = new HashMap<>(); for (TopicPartition topicPartition : consumer.assignment()) { positions.put(topicPartition, new OffsetAndMetadata(consumer.position(topicPartition), null)); } + // Checkpoint the progress by sending offsets to group coordinator broker. producer.sendOffsetsToTransaction(positions, consumer.groupMetadata()); + // Finish the transaction. producer.commitTransaction(); } catch (CommitFailedException e) { producer.abortTransaction(); @@ -58,11 +92,24 @@ public static void main(String[] args) { throw new KafkaException("Encountered fatal error during processing: " + e.getMessage()); } } + messageRemaining = messagesRemaining(consumer); + System.out.println("Message remaining: " + messageRemaining); } } - private static ConsumerRecord transform(ConsumerRecord record) { + private static ProducerRecord transform(ConsumerRecord record) { // Customized business logic here - return record; + return new ProducerRecord<>(OUTPUT_TOPIC, record.key() / 2, record.value()); + } + + private static long messagesRemaining(KafkaConsumer consumer) { + return consumer.assignment().stream().mapToLong(partition -> { + long currentPosition = consumer.position(partition); + Map endOffsets = consumer.endOffsets(singleton(partition)); + if (endOffsets.containsKey(partition)) { + return endOffsets.get(partition) - currentPosition; + } + return 0; + }).sum(); } } diff --git a/examples/src/main/java/kafka/examples/Producer.java b/examples/src/main/java/kafka/examples/Producer.java index e37f4cd9fe160..8ffd419d37b82 100644 --- a/examples/src/main/java/kafka/examples/Producer.java +++ b/examples/src/main/java/kafka/examples/Producer.java @@ -51,7 +51,7 @@ KafkaProducer get() { } public void run() { - int messageNo = 1; + int messageNo = 0; while (true) { String messageStr = "Message_" + messageNo; long startTime = System.currentTimeMillis(); @@ -69,7 +69,7 @@ public void run() { e.printStackTrace(); } } - ++messageNo; + messageNo += 2; } } } From b038d7fa18760f50c87cd07f87336f9a90084037 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Fri, 31 Jan 2020 19:24:25 -0500 Subject: [PATCH 03/11] build entire java class with workflow --- examples/README | 4 + examples/bin/exactly-once-demo.sh | 23 +++ .../examples/ExactlyOnceMessageProcessor.java | 170 ++++++++++++++++++ .../examples/KafkaConsumerProducerDemo.java | 5 +- .../kafka/examples/KafkaExactlyOnceDemo.java | 137 +++++--------- .../main/java/kafka/examples/Producer.java | 39 ++-- 6 files changed, 262 insertions(+), 116 deletions(-) create mode 100755 examples/bin/exactly-once-demo.sh create mode 100644 examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java diff --git a/examples/README b/examples/README index 887cbd6b76f72..2bdba108352e7 100644 --- a/examples/README +++ b/examples/README @@ -6,3 +6,7 @@ To run the demo: 2. For simple consumer demo, `run bin/java-simple-consumer-demo.sh` 3. For unlimited sync-producer-consumer run, `run bin/java-producer-consumer-demo.sh sync` 4. For unlimited async-producer-consumer run, `run bin/java-producer-consumer-demo.sh` + 5. For standalone mode exactly once demo run, `run bin/exactly-once-demo.sh standaloneMode 6 3 100000`, + this means we are starting 3 EOS instances with 6 topic partitions and 100000 pre-populated records + 6. For group mode exactly once demo run, `run bin/exactly-once-demo.sh groupMode 6 3 100000`, + this means the same as the standalone demo, except consumers are using subscription mode \ No newline at end of file diff --git a/examples/bin/exactly-once-demo.sh b/examples/bin/exactly-once-demo.sh new file mode 100755 index 0000000000000..74db420816471 --- /dev/null +++ b/examples/bin/exactly-once-demo.sh @@ -0,0 +1,23 @@ +#!/bin/bash +# 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. + +base_dir=$(dirname $0)/../.. + +if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then + export KAFKA_HEAP_OPTS="-Xmx512M" +fi + +exec $base_dir/bin/kafka-run-class.sh kafka.examples.KafkaConsumerProducerDemo $@ diff --git a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java new file mode 100644 index 0000000000000..a8e2215fb434e --- /dev/null +++ b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java @@ -0,0 +1,170 @@ +/* + * 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.consumer.CommitFailedException; +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.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 org.apache.kafka.common.errors.FencedInstanceIdException; +import org.apache.kafka.common.errors.ProducerFencedException; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; + +import static java.util.Collections.singleton; + +/** + * A demo class for how to write a customized EOS app. It takes a consume-process-produce loop + * The things to pay attention to beyond a general consumer + producer app are: + * 1. Define a unique transactional.id for your app + * 2. Turn on read_committed isolation level + */ +public class ExactlyOnceMessageProcessor extends Thread { + + private static final boolean READ_COMMITTED = true; + + private final String mode; + private final String inputTopic; + private final String outputTopic; + private final int numPartitions; + private final int numInstances; + private final int instanceIdx; + private final String transactionalId; + + private final KafkaProducer producer; + private final KafkaConsumer consumer; + + public ExactlyOnceMessageProcessor(final String mode, + final String inputTopic, + final String outputTopic, + final int numPartitions, + final int numInstances, + final int instanceIdx) { + this.mode = mode; + this.inputTopic = inputTopic; + this.outputTopic = outputTopic; + this.numPartitions = numPartitions; + this.numInstances = numInstances; + this.instanceIdx = instanceIdx; + this.transactionalId = "Processor-" + instanceIdx; + producer = new Producer(outputTopic, true, transactionalId, -1).get(); + consumer = new Consumer(inputTopic, READ_COMMITTED).get(); + } + + @Override + public void run() { + // Init transactions call should always happen first in order to clear zombie transactions. + producer.initTransactions(); + + final AtomicLong messageRemaining = new AtomicLong(Long.MAX_VALUE); + + // Under group mode, topic based subscription is sufficient as Consumers are safe to work transactionally after 2.5. + // Under standalone mode, user needs to manually assign the topic partitions and make sure the assignment is unique + // across the consumer group. + if (this.mode.equals("groupMode")) { + consumer.subscribe(Collections.singleton(KafkaProperties.TOPIC), new ConsumerRebalanceListener() { + @Override + public void onPartitionsRevoked(Collection partitions) { + printWithPrefix("Revoked partition assignment to kick-off rebalancing: " + partitions); + } + + @Override + public void onPartitionsAssigned(Collection partitions) { + printWithPrefix("Received partition assignment after rebalancing: " + partitions); + messageRemaining.set(messagesRemaining(consumer)); + } + }); + } else { + List topicPartitions = new ArrayList<>(); + int rangeSize = numPartitions / numInstances; + int startPartition = rangeSize * instanceIdx; + int endPartition = Math.min(numPartitions - 1, startPartition + rangeSize - 1); + for (int partition = startPartition; partition <= endPartition; partition++) { + topicPartitions.add(new TopicPartition(inputTopic, partition)); + } + + consumer.assign(topicPartitions); + printWithPrefix("Manually assign partitions: " + topicPartitions); + } + + int messageProcessed = 0; + while (messageRemaining.get() > 0) { + ConsumerRecords records = consumer.poll(Duration.ofMillis(200)); + if (records.count() > 0) { + try { + // Begin a new transaction session. + producer.beginTransaction(); + for (ConsumerRecord record : records) { + ProducerRecord customizedRecord = transform(record); + // Send records to the downstream. + producer.send(customizedRecord); + } + Map positions = new HashMap<>(); + for (TopicPartition topicPartition : consumer.assignment()) { + positions.put(topicPartition, new OffsetAndMetadata(consumer.position(topicPartition), null)); + } + // Checkpoint the progress by sending offsets to group coordinator broker. + producer.sendOffsetsToTransaction(positions, consumer.groupMetadata()); + // Finish the transaction. + producer.commitTransaction(); + messageProcessed += records.count(); + } catch (CommitFailedException e) { + producer.abortTransaction(); + } catch (ProducerFencedException | FencedInstanceIdException e) { + throw new KafkaException("Encountered fatal error during processing: " + e.getMessage()); + } + } + messageRemaining.set(messagesRemaining(consumer)); + printWithPrefix("Message remaining: " + messageRemaining); + } + + printWithPrefix("Finished processing " + messageProcessed + " records"); + } + + private void printWithPrefix(String message) { + System.out.println(transactionalId + ": " + message); + } + + private ProducerRecord transform(ConsumerRecord record) { + // Customized business logic here + return new ProducerRecord<>(outputTopic, record.key() / 2, record.value()); + } + + private static long messagesRemaining(KafkaConsumer consumer) { + return consumer.assignment().stream().mapToLong(partition -> { + long currentPosition = consumer.position(partition); + Map endOffsets = consumer.endOffsets(singleton(partition)); + if (endOffsets.containsKey(partition)) { + return endOffsets.get(partition) - currentPosition; + } + return 0; + }).sum(); + } +} diff --git a/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java b/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java index ba68fd16c4fdd..6e8686c8ff644 100644 --- a/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java +++ b/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java @@ -19,11 +19,10 @@ public class KafkaConsumerProducerDemo { public static void main(String[] args) { boolean isAsync = args.length == 0 || !args[0].trim().equalsIgnoreCase("sync"); - Producer producerThread = new Producer(KafkaProperties.TOPIC, isAsync, null); + Producer producerThread = new Producer(KafkaProperties.TOPIC, isAsync, null, 10000); producerThread.start(); - Consumer consumerThread = new Consumer(KafkaProperties.TOPIC); + Consumer consumerThread = new Consumer(KafkaProperties.TOPIC, false); consumerThread.start(); - } } diff --git a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java index d60e05ba086f7..b60a97fc1af06 100644 --- a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java +++ b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java @@ -1,115 +1,58 @@ +/* + * 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.consumer.CommitFailedException; -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.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 org.apache.kafka.common.errors.FencedInstanceIdException; -import org.apache.kafka.common.errors.ProducerFencedException; - -import java.time.Duration; -import java.util.Collection; -import java.util.Collections; -import java.util.HashMap; -import java.util.Map; -import java.util.UUID; - -import static java.util.Collections.singleton; +import java.util.ArrayList; +import java.util.List; -/** - * A demo class for how to write a customized EOS app. It takes a consume-process-produce loop - * The things to pay attention to beyond a general consumer + producer app are: - * 1. Define a unique transactional.id for your app - * 2. Turn on read_committed isolation level - */ public class KafkaExactlyOnceDemo { + private static final String INPUT_TOPIC = "input-topic"; private static final String OUTPUT_TOPIC = "output-topic"; - private static final String TRANSACTIONAL_ID = "transactional-id-" + UUID.randomUUID(); - private static final boolean READ_COMMITTED = true; - public static void main(String[] args) { - if (args.length < 1) { - throw new IllegalArgumentException("Should accept at least one parameter"); - } else if (args[0].equals("standaloneMode") && args.length != 2) { - throw new IllegalArgumentException("Should have specified partition in standalone mode"); + public static void main(String[] args) throws InterruptedException { + if (args.length != 4) { + throw new IllegalArgumentException("Should accept 4 parameters: [mode], " + + "[number of partitions], [number of instances], [number of records]"); } - KafkaProducer producer = new Producer(KafkaProperties.TOPIC, true, TRANSACTIONAL_ID).get(); - // Init transactions call should always happen first in order to clear zombie transactions. - producer.initTransactions(); - - KafkaConsumer consumer = new Consumer(KafkaProperties.TOPIC, READ_COMMITTED).get(); - // Under group mode, topic based subscription is sufficient as Consumers are safe to work transactionally after 2.5. - // Under standalone mode, user needs to manually assign the topic partitions and make sure the assignment is unique - // across the consumer group. - if (args[0].equals("groupMode")) { - consumer.subscribe(Collections.singleton(KafkaProperties.TOPIC), new ConsumerRebalanceListener() { - @Override - public void onPartitionsRevoked(Collection partitions) { + String mode = args[0]; + int numPartitions = Integer.valueOf(args[1]); + int numInstances = Integer.valueOf(args[2]); + int numRecords = Integer.valueOf(args[3]); - } + // Pre-populate records + Producer producerThread = new Producer(INPUT_TOPIC, true, null, numRecords); + producerThread.start(); - @Override - public void onPartitionsAssigned(Collection partitions) { - System.out.println("Get partition assignment " + partitions); - } - }); - } else { - consumer.assign(Collections.singletonList(new TopicPartition(KafkaProperties.TOPIC, - Integer.valueOf(args[1])))); + synchronized (producerThread.getClass()) { + while (producerThread.isAlive()) { + producerThread.getClass().wait(); + } } - long messageRemaining = messagesRemaining(consumer); - while (messageRemaining > 0) { - ConsumerRecords records = consumer.poll(Duration.ofMillis(200)); - if (records.count() > 0) { - try { - // Begin a new transaction session. - producer.beginTransaction(); - for (ConsumerRecord record : records) { - ProducerRecord customizedRecord = transform(record); - // Send records to the downstream. - producer.send(customizedRecord); - } - Map positions = new HashMap<>(); - for (TopicPartition topicPartition : consumer.assignment()) { - positions.put(topicPartition, new OffsetAndMetadata(consumer.position(topicPartition), null)); - } - // Checkpoint the progress by sending offsets to group coordinator broker. - producer.sendOffsetsToTransaction(positions, consumer.groupMetadata()); - // Finish the transaction. - producer.commitTransaction(); - } catch (CommitFailedException e) { - producer.abortTransaction(); - } catch (ProducerFencedException | FencedInstanceIdException e) { - throw new KafkaException("Encountered fatal error during processing: " + e.getMessage()); - } - } - messageRemaining = messagesRemaining(consumer); - System.out.println("Message remaining: " + messageRemaining); + List eosProcessors = new ArrayList<>(numInstances); + for (int instanceIdx = 0; instanceIdx < numInstances; instanceIdx++) { + ExactlyOnceMessageProcessor messageProcessor = new ExactlyOnceMessageProcessor(mode, + INPUT_TOPIC, OUTPUT_TOPIC, numPartitions, numInstances, instanceIdx); + eosProcessors.add(messageProcessor); + messageProcessor.start(); } - } - private static ProducerRecord transform(ConsumerRecord record) { - // Customized business logic here - return new ProducerRecord<>(OUTPUT_TOPIC, record.key() / 2, record.value()); } - private static long messagesRemaining(KafkaConsumer consumer) { - return consumer.assignment().stream().mapToLong(partition -> { - long currentPosition = consumer.position(partition); - Map endOffsets = consumer.endOffsets(singleton(partition)); - if (endOffsets.containsKey(partition)) { - return endOffsets.get(partition) - currentPosition; - } - return 0; - }).sum(); - } } diff --git a/examples/src/main/java/kafka/examples/Producer.java b/examples/src/main/java/kafka/examples/Producer.java index 8ffd419d37b82..4afc9834d68a5 100644 --- a/examples/src/main/java/kafka/examples/Producer.java +++ b/examples/src/main/java/kafka/examples/Producer.java @@ -31,8 +31,9 @@ public class Producer extends Thread { private final KafkaProducer producer; private final String topic; private final Boolean isAsync; + private int numRecords; - public Producer(String topic, Boolean isAsync, String transactionalId) { + public Producer(String topic, Boolean isAsync, String transactionalId, int numRecords) { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT); props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoProducer"); @@ -44,32 +45,38 @@ public Producer(String topic, Boolean isAsync, String transactionalId) { producer = new KafkaProducer<>(props); this.topic = topic; this.isAsync = isAsync; + this.numRecords = numRecords; } KafkaProducer get() { return producer; } + @Override public void run() { - int messageNo = 0; - while (true) { - String messageStr = "Message_" + messageNo; - long startTime = System.currentTimeMillis(); - if (isAsync) { // Send asynchronously - producer.send(new ProducerRecord<>(topic, - messageNo, - messageStr), new DemoCallBack(startTime, messageNo, messageStr)); - } else { // Send synchronously - try { + synchronized (this) { + int messageNo = 0; + while (numRecords > 0) { + String messageStr = "Message_" + messageNo; + long startTime = System.currentTimeMillis(); + if (isAsync) { // Send asynchronously producer.send(new ProducerRecord<>(topic, messageNo, - messageStr)).get(); - System.out.println("Sent message: (" + messageNo + ", " + messageStr + ")"); - } catch (InterruptedException | ExecutionException e) { - e.printStackTrace(); + messageStr), new DemoCallBack(startTime, messageNo, messageStr)); + } else { // Send synchronously + try { + producer.send(new ProducerRecord<>(topic, + messageNo, + messageStr)).get(); + System.out.println("Sent message: (" + messageNo + ", " + messageStr + ")"); + } catch (InterruptedException | ExecutionException e) { + e.printStackTrace(); + } } + messageNo += 2; + numRecords -= 1; } - messageNo += 2; + this.notifyAll(); } } } From bfc1b93c24dc17687456ebfce5adee5282751aeb Mon Sep 17 00:00:00 2001 From: abbccdda Date: Fri, 31 Jan 2020 21:22:44 -0600 Subject: [PATCH 04/11] create topic --- examples/bin/exactly-once-demo.sh | 2 +- .../main/java/kafka/examples/Consumer.java | 17 +++++- .../examples/ExactlyOnceMessageProcessor.java | 12 +++- .../examples/KafkaConsumerProducerDemo.java | 15 +++-- .../kafka/examples/KafkaExactlyOnceDemo.java | 56 +++++++++++++++---- .../main/java/kafka/examples/Producer.java | 41 +++++++------- 6 files changed, 101 insertions(+), 42 deletions(-) diff --git a/examples/bin/exactly-once-demo.sh b/examples/bin/exactly-once-demo.sh index 74db420816471..e9faa428d3f27 100755 --- a/examples/bin/exactly-once-demo.sh +++ b/examples/bin/exactly-once-demo.sh @@ -20,4 +20,4 @@ if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then export KAFKA_HEAP_OPTS="-Xmx512M" fi -exec $base_dir/bin/kafka-run-class.sh kafka.examples.KafkaConsumerProducerDemo $@ +exec $base_dir/bin/kafka-run-class.sh kafka.examples.KafkaExactlyOnceDemo $@ diff --git a/examples/src/main/java/kafka/examples/Consumer.java b/examples/src/main/java/kafka/examples/Consumer.java index 6556d27008b92..05def28d13d7b 100644 --- a/examples/src/main/java/kafka/examples/Consumer.java +++ b/examples/src/main/java/kafka/examples/Consumer.java @@ -26,12 +26,19 @@ import java.time.Duration; import java.util.Collections; import java.util.Properties; +import java.util.concurrent.CountDownLatch; public class Consumer extends ShutdownableThread { private final KafkaConsumer consumer; private final String topic; + private final int numMessageToConsume; + private int messageRemaining; + private final CountDownLatch latch; - public Consumer(String topic, boolean readCommitted) { + public Consumer(final String topic, + final boolean readCommitted, + final int numMessageToConsume, + final CountDownLatch latch) { super("KafkaConsumerExample", false); Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT); @@ -47,6 +54,9 @@ public Consumer(String topic, boolean readCommitted) { consumer = new KafkaConsumer<>(props); this.topic = topic; + this.numMessageToConsume = numMessageToConsume; + this.messageRemaining = numMessageToConsume; + this.latch = latch; } KafkaConsumer get() { @@ -60,6 +70,11 @@ public void doWork() { for (ConsumerRecord record : records) { System.out.println("Received message: (" + record.key() + ", " + record.value() + ") at offset " + record.offset()); } + messageRemaining -= records.count(); + if (messageRemaining <= 0) { + System.out.println("Read committed consumer finished reading " + numMessageToConsume + " messages"); + } + latch.countDown(); } @Override diff --git a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java index a8e2215fb434e..3ad4bbd5b8476 100644 --- a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java +++ b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java @@ -36,6 +36,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicLong; import static java.util.Collections.singleton; @@ -61,12 +62,15 @@ public class ExactlyOnceMessageProcessor extends Thread { private final KafkaProducer producer; private final KafkaConsumer consumer; + private final CountDownLatch latch; + public ExactlyOnceMessageProcessor(final String mode, final String inputTopic, final String outputTopic, final int numPartitions, final int numInstances, - final int instanceIdx) { + final int instanceIdx, + final CountDownLatch latch) { this.mode = mode; this.inputTopic = inputTopic; this.outputTopic = outputTopic; @@ -74,8 +78,9 @@ public ExactlyOnceMessageProcessor(final String mode, this.numInstances = numInstances; this.instanceIdx = instanceIdx; this.transactionalId = "Processor-" + instanceIdx; - producer = new Producer(outputTopic, true, transactionalId, -1).get(); - consumer = new Consumer(inputTopic, READ_COMMITTED).get(); + producer = new Producer(outputTopic, true, transactionalId, -1, null).get(); + consumer = new Consumer(inputTopic, READ_COMMITTED, -1, null).get(); + this.latch = latch; } @Override @@ -146,6 +151,7 @@ public void onPartitionsAssigned(Collection partitions) { } printWithPrefix("Finished processing " + messageProcessed + " records"); + latch.countDown(); } private void printWithPrefix(String message) { diff --git a/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java b/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java index 6e8686c8ff644..db1c25e7f7597 100644 --- a/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java +++ b/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java @@ -16,13 +16,18 @@ */ package kafka.examples; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + public class KafkaConsumerProducerDemo { - public static void main(String[] args) { + public static void main(String[] args) throws InterruptedException { boolean isAsync = args.length == 0 || !args[0].trim().equalsIgnoreCase("sync"); - Producer producerThread = new Producer(KafkaProperties.TOPIC, isAsync, null, 10000); - producerThread.start(); - + CountDownLatch latch = new CountDownLatch(1); + Producer producerThread = new Producer(KafkaProperties.TOPIC, isAsync, null, 10000, latch); + producerThread.start(); + Consumer consumerThread = new Consumer(KafkaProperties.TOPIC, false); - consumerThread.start(); + consumerThread.start(); + 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 b60a97fc1af06..29b51871f0d47 100644 --- a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java +++ b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java @@ -16,15 +16,23 @@ */ package kafka.examples; -import java.util.ArrayList; -import java.util.List; +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.CreateTopicsResult; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.consumer.ConsumerConfig; + +import java.util.Arrays; +import java.util.Properties; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; public class KafkaExactlyOnceDemo { private static final String INPUT_TOPIC = "input-topic"; private static final String OUTPUT_TOPIC = "output-topic"; - public static void main(String[] args) throws InterruptedException { + public static void main(String[] args) throws InterruptedException, ExecutionException { if (args.length != 4) { throw new IllegalArgumentException("Should accept 4 parameters: [mode], " + "[number of partitions], [number of instances], [number of records]"); @@ -35,24 +43,48 @@ public static void main(String[] args) throws InterruptedException { int numInstances = Integer.valueOf(args[2]); int numRecords = Integer.valueOf(args[3]); + createTopics(numPartitions); + + CountDownLatch prePopulateLatch = new CountDownLatch(1); + // Pre-populate records - Producer producerThread = new Producer(INPUT_TOPIC, true, null, numRecords); + Producer producerThread = new Producer(INPUT_TOPIC, true, null, numRecords, prePopulateLatch); producerThread.start(); - synchronized (producerThread.getClass()) { - while (producerThread.isAlive()) { - producerThread.getClass().wait(); - } - } + prePopulateLatch.await(5, TimeUnit.MINUTES); + + CountDownLatch transactionalCopyLatch = new CountDownLatch(numInstances); - List eosProcessors = new ArrayList<>(numInstances); + // Transactionally copy over all messages for (int instanceIdx = 0; instanceIdx < numInstances; instanceIdx++) { ExactlyOnceMessageProcessor messageProcessor = new ExactlyOnceMessageProcessor(mode, - INPUT_TOPIC, OUTPUT_TOPIC, numPartitions, numInstances, instanceIdx); - eosProcessors.add(messageProcessor); + INPUT_TOPIC, OUTPUT_TOPIC, numPartitions, + numInstances, instanceIdx, transactionalCopyLatch); messageProcessor.start(); } + transactionalCopyLatch.await(5, TimeUnit.MINUTES); + + CountDownLatch consumeLatch = new CountDownLatch(1); + + Consumer consumerThread = new Consumer(KafkaProperties.TOPIC, true, numRecords, consumeLatch); + consumerThread.start(); + + consumeLatch.await(5, TimeUnit.MINUTES); + consumerThread.shutdown(); + System.out.println("All finished!"); } + private static void createTopics(final int numPartitions) throws ExecutionException, InterruptedException { + Properties properties = new Properties(); + properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); + + Admin adminClient = Admin.create(properties); + short replicationFactor = 1; + CreateTopicsResult result = adminClient.createTopics(Arrays.asList( + new NewTopic(INPUT_TOPIC, numPartitions, replicationFactor), + new NewTopic(OUTPUT_TOPIC, numPartitions, replicationFactor))); + + result.all().get(); + } } diff --git a/examples/src/main/java/kafka/examples/Producer.java b/examples/src/main/java/kafka/examples/Producer.java index 4afc9834d68a5..ae996a3c74907 100644 --- a/examples/src/main/java/kafka/examples/Producer.java +++ b/examples/src/main/java/kafka/examples/Producer.java @@ -25,6 +25,7 @@ import org.apache.kafka.common.serialization.StringSerializer; import java.util.Properties; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; public class Producer extends Thread { @@ -32,8 +33,9 @@ public class Producer extends Thread { private final String topic; private final Boolean isAsync; private int numRecords; + private final CountDownLatch latch; - public Producer(String topic, Boolean isAsync, String transactionalId, int numRecords) { + public Producer(String topic, Boolean isAsync, String transactionalId, int numRecords, CountDownLatch latch) { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT); props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoProducer"); @@ -46,6 +48,7 @@ public Producer(String topic, Boolean isAsync, String transactionalId, int numRe this.topic = topic; this.isAsync = isAsync; this.numRecords = numRecords; + this.latch = latch; } KafkaProducer get() { @@ -54,30 +57,28 @@ KafkaProducer get() { @Override public void run() { - synchronized (this) { - int messageNo = 0; - while (numRecords > 0) { - String messageStr = "Message_" + messageNo; - long startTime = System.currentTimeMillis(); - if (isAsync) { // Send asynchronously + int messageNo = 0; + while (numRecords > 0) { + String messageStr = "Message_" + messageNo; + long startTime = System.currentTimeMillis(); + if (isAsync) { // Send asynchronously + producer.send(new ProducerRecord<>(topic, + messageNo, + messageStr), new DemoCallBack(startTime, messageNo, messageStr)); + } else { // Send synchronously + try { producer.send(new ProducerRecord<>(topic, messageNo, - messageStr), new DemoCallBack(startTime, messageNo, messageStr)); - } else { // Send synchronously - try { - producer.send(new ProducerRecord<>(topic, - messageNo, - messageStr)).get(); - System.out.println("Sent message: (" + messageNo + ", " + messageStr + ")"); - } catch (InterruptedException | ExecutionException e) { - e.printStackTrace(); - } + messageStr)).get(); + System.out.println("Sent message: (" + messageNo + ", " + messageStr + ")"); + } catch (InterruptedException | ExecutionException e) { + e.printStackTrace(); } - messageNo += 2; - numRecords -= 1; } - this.notifyAll(); + messageNo += 2; + numRecords -= 1; } + latch.countDown(); } } From 239e3bb794c978d54ead2578f02c4fc5be7cdd91 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Fri, 31 Jan 2020 23:28:00 -0600 Subject: [PATCH 05/11] Only topic clean-up remaining --- .../clients/admin/DescribeTopicsResult.java | 23 ++-- .../kafka/clients/admin/ListTopicsResult.java | 14 +-- .../main/java/kafka/examples/Consumer.java | 10 +- .../examples/ExactlyOnceMessageProcessor.java | 25 ++-- .../examples/KafkaConsumerProducerDemo.java | 6 +- .../kafka/examples/KafkaExactlyOnceDemo.java | 107 +++++++++++++++--- .../main/java/kafka/examples/Producer.java | 15 ++- 7 files changed, 140 insertions(+), 60 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTopicsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTopicsResult.java index 9822b423c4d82..ffb6470e503c5 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTopicsResult.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTopicsResult.java @@ -51,21 +51,18 @@ public Map> values() { */ public KafkaFuture> all() { return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0])). - thenApply(new KafkaFuture.BaseFunction>() { - @Override - public Map apply(Void v) { - Map descriptions = new HashMap<>(futures.size()); - for (Map.Entry> entry : futures.entrySet()) { - try { - descriptions.put(entry.getKey(), entry.getValue().get()); - } catch (InterruptedException | ExecutionException e) { - // This should be unreachable, because allOf ensured that all the futures - // completed successfully. - throw new RuntimeException(e); - } + thenApply(v -> { + Map descriptions = new HashMap<>(futures.size()); + for (Map.Entry> entry : futures.entrySet()) { + try { + descriptions.put(entry.getKey(), entry.getValue().get()); + } catch (InterruptedException | ExecutionException e) { + // This should be unreachable, because allOf ensured that all the futures + // completed successfully. + throw new RuntimeException(e); } - return descriptions; } + return descriptions; }); } } diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsResult.java index 4e7e1a2e98721..21540732d30d2 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsResult.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsResult.java @@ -48,23 +48,13 @@ public KafkaFuture> namesToListings() { * Return a future which yields a collection of TopicListing objects. */ public KafkaFuture> listings() { - return future.thenApply(new KafkaFuture.BaseFunction, Collection>() { - @Override - public Collection apply(Map namesToDescriptions) { - return namesToDescriptions.values(); - } - }); + return future.thenApply(namesToDescriptions -> namesToDescriptions.values()); } /** * Return a future which yields a collection of topic names. */ public KafkaFuture> names() { - return future.thenApply(new KafkaFuture.BaseFunction, Set>() { - @Override - public Set apply(Map namesToListings) { - return namesToListings.keySet(); - } - }); + return future.thenApply(namesToListings -> namesToListings.keySet()); } } diff --git a/examples/src/main/java/kafka/examples/Consumer.java b/examples/src/main/java/kafka/examples/Consumer.java index 05def28d13d7b..df0a884539287 100644 --- a/examples/src/main/java/kafka/examples/Consumer.java +++ b/examples/src/main/java/kafka/examples/Consumer.java @@ -36,13 +36,14 @@ public class Consumer extends ShutdownableThread { private final CountDownLatch latch; public Consumer(final String topic, + final String groupId, final boolean readCommitted, final int numMessageToConsume, final CountDownLatch latch) { super("KafkaConsumerExample", false); Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT); - props.put(ConsumerConfig.GROUP_ID_CONFIG, "DemoConsumer"); + props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000"); @@ -51,6 +52,7 @@ public Consumer(final String topic, if (readCommitted) { props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"); } + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); consumer = new KafkaConsumer<>(props); this.topic = topic; @@ -68,13 +70,13 @@ public void doWork() { consumer.subscribe(Collections.singletonList(this.topic)); ConsumerRecords records = consumer.poll(Duration.ofSeconds(1)); for (ConsumerRecord record : records) { - System.out.println("Received message: (" + record.key() + ", " + record.value() + ") at offset " + record.offset()); + System.out.println("Verify-consumer received message : from partition " + record.partition() + ", (" + record.key() + ", " + record.value() + ") at offset " + record.offset()); } messageRemaining -= records.count(); if (messageRemaining <= 0) { - System.out.println("Read committed consumer finished reading " + numMessageToConsume + " messages"); + System.out.println("Verify-consumer finished reading " + numMessageToConsume + " messages"); + latch.countDown(); } - latch.countDown(); } @Override diff --git a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java index 3ad4bbd5b8476..be9ae2720a223 100644 --- a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java +++ b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java @@ -28,6 +28,7 @@ import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.FencedInstanceIdException; import org.apache.kafka.common.errors.ProducerFencedException; +import org.apache.kafka.common.internals.Topic; import java.time.Duration; import java.util.ArrayList; @@ -39,8 +40,6 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicLong; -import static java.util.Collections.singleton; - /** * A demo class for how to write a customized EOS app. It takes a consume-process-produce loop * The things to pay attention to beyond a general consumer + producer app are: @@ -78,8 +77,8 @@ public ExactlyOnceMessageProcessor(final String mode, this.numInstances = numInstances; this.instanceIdx = instanceIdx; this.transactionalId = "Processor-" + instanceIdx; - producer = new Producer(outputTopic, true, transactionalId, -1, null).get(); - consumer = new Consumer(inputTopic, READ_COMMITTED, -1, null).get(); + producer = new Producer(outputTopic, true, transactionalId, true, -1, null).get(); + consumer = new Consumer(inputTopic, "Eos-consumer", READ_COMMITTED, -1, null).get(); this.latch = latch; } @@ -159,16 +158,22 @@ private void printWithPrefix(String message) { } private ProducerRecord transform(ConsumerRecord record) { - // Customized business logic here - return new ProducerRecord<>(outputTopic, record.key() / 2, record.value()); + printWithPrefix("Transformed record (" + record.key() + "," + record.value() + ")"); + return new ProducerRecord<>(outputTopic, record.key() / 2, "Transformed_" + record.value()); } - private static long messagesRemaining(KafkaConsumer consumer) { + private long messagesRemaining(KafkaConsumer consumer) { + // If we couldn't detect any end offset, that means we are still not able to fetch offsets. + final Map fullEndOffsets = consumer.endOffsets(new ArrayList<>(consumer.assignment())); + if (fullEndOffsets.isEmpty()) { + return Long.MAX_VALUE; + } + return consumer.assignment().stream().mapToLong(partition -> { long currentPosition = consumer.position(partition); - Map endOffsets = consumer.endOffsets(singleton(partition)); - if (endOffsets.containsKey(partition)) { - return endOffsets.get(partition) - currentPosition; + printWithPrefix("Processing partition " + partition + " with full offsets " + fullEndOffsets); + if (fullEndOffsets.containsKey(partition)) { + return fullEndOffsets.get(partition) - currentPosition; } return 0; }).sum(); diff --git a/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java b/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java index db1c25e7f7597..9a4a50923971c 100644 --- a/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java +++ b/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java @@ -22,11 +22,11 @@ public class KafkaConsumerProducerDemo { public static void main(String[] args) throws InterruptedException { boolean isAsync = args.length == 0 || !args[0].trim().equalsIgnoreCase("sync"); - CountDownLatch latch = new CountDownLatch(1); - Producer producerThread = new Producer(KafkaProperties.TOPIC, isAsync, null, 10000, latch); + CountDownLatch latch = new CountDownLatch(2); + Producer producerThread = new Producer(KafkaProperties.TOPIC, isAsync, null, false, 10000, latch); producerThread.start(); - Consumer consumerThread = new Consumer(KafkaProperties.TOPIC, false); + Consumer consumerThread = new Consumer(KafkaProperties.TOPIC, "DemoConsumer", false, 10000, latch); consumerThread.start(); 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 29b51871f0d47..2bfd5701ce00d 100644 --- a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java +++ b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java @@ -17,16 +17,29 @@ package kafka.examples; import org.apache.kafka.clients.admin.Admin; -import org.apache.kafka.clients.admin.CreateTopicsResult; +import org.apache.kafka.clients.admin.DescribeTopicsResult; +import org.apache.kafka.clients.admin.ListTopicsResult; import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.admin.TopicDescription; import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.KafkaFuture; +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.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +/** + * This exactly once demo driver is following below steps: + * 1. Set up a producer to pre populate a set of records into input topic + * 2. + */ public class KafkaExactlyOnceDemo { private static final String INPUT_TOPIC = "input-topic"; @@ -43,19 +56,21 @@ public static void main(String[] args) throws InterruptedException, ExecutionExc int numInstances = Integer.valueOf(args[2]); int numRecords = Integer.valueOf(args[3]); - createTopics(numPartitions); + recreateTopics(numPartitions); CountDownLatch prePopulateLatch = new CountDownLatch(1); - // Pre-populate records - Producer producerThread = new Producer(INPUT_TOPIC, true, null, numRecords, prePopulateLatch); + // Pre-populate records. + final boolean isAsync = false; + final boolean enableIdempotency = true; + Producer producerThread = new Producer(INPUT_TOPIC, isAsync, null, enableIdempotency, numRecords, prePopulateLatch); producerThread.start(); prePopulateLatch.await(5, TimeUnit.MINUTES); CountDownLatch transactionalCopyLatch = new CountDownLatch(numInstances); - // Transactionally copy over all messages + // Transactionally copy over all messages. for (int instanceIdx = 0; instanceIdx < numInstances; instanceIdx++) { ExactlyOnceMessageProcessor messageProcessor = new ExactlyOnceMessageProcessor(mode, INPUT_TOPIC, OUTPUT_TOPIC, numPartitions, @@ -67,7 +82,7 @@ public static void main(String[] args) throws InterruptedException, ExecutionExc CountDownLatch consumeLatch = new CountDownLatch(1); - Consumer consumerThread = new Consumer(KafkaProperties.TOPIC, true, numRecords, consumeLatch); + Consumer consumerThread = new Consumer(OUTPUT_TOPIC, "Verify-consumer", true, numRecords, consumeLatch); consumerThread.start(); consumeLatch.await(5, TimeUnit.MINUTES); @@ -75,16 +90,78 @@ public static void main(String[] args) throws InterruptedException, ExecutionExc System.out.println("All finished!"); } - private static void createTopics(final int numPartitions) throws ExecutionException, InterruptedException { - Properties properties = new Properties(); - properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); + private static void recreateTopics(final int numPartitions) throws ExecutionException, InterruptedException { + Properties props = new Properties(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT); + + Admin adminClient = Admin.create(props); - Admin adminClient = Admin.create(properties); - short replicationFactor = 1; - CreateTopicsResult result = adminClient.createTopics(Arrays.asList( - new NewTopic(INPUT_TOPIC, numPartitions, replicationFactor), - new NewTopic(OUTPUT_TOPIC, numPartitions, replicationFactor))); + List topicsToDelete = Arrays.asList(INPUT_TOPIC, OUTPUT_TOPIC); + + try { + adminClient.deleteTopics(topicsToDelete).all().get(); + } catch (ExecutionException e) { + if (!(e.getCause() instanceof UnknownTopicOrPartitionException)) { + throw e; + } + System.out.println("Encountered exception during topic deletion: " + e.getCause()); + } + System.out.println("Deleted old topics: " + topicsToDelete); + + // Check topic existence in a retry loop + while (true) { + System.out.println("Making sure the topics are deleted successfully: " + topicsToDelete); + boolean noTopicsInfo = true; + + Set listedTopics = adminClient.listTopics().names().get(); + System.out.println("Current list of topics: " + listedTopics); + + for (String topic : topicsToDelete) { + System.out.println("Checking topic " + topic); + if (listedTopics.contains(topic)) { + noTopicsInfo = false; + break; + } + } + + if (noTopicsInfo) { + break; + } + Thread.sleep(1000); + } - result.all().get(); + // Create topics in a retry loop + while (true) { + final short replicationFactor = 1; + final List newTopics = Arrays.asList( + new NewTopic(INPUT_TOPIC, numPartitions, replicationFactor), + new NewTopic(OUTPUT_TOPIC, numPartitions, replicationFactor)); + try { + adminClient.createTopics(newTopics).all().get(); + System.out.println("Created new topics: " + newTopics); + break; + } catch (ExecutionException e) { + if (!(e.getCause() instanceof TopicExistsException)) { + throw e; + } + System.out.println("Metadata of the old topics are not cleared yet..."); + Thread.sleep(1000); + } + } } } + +// DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(topicsToDelete); +// for (KafkaFuture future : describeTopicsResult.values().values()) { +// try { +// TopicDescription description = future.get(); +// System.out.println("Found topic description: " + description); +// noTopicsInfo = false; +// break; +// } catch (ExecutionException e) { +// if (!(e.getCause() instanceof UnknownTopicOrPartitionException)) { +// throw e; +// } +// } +// } + diff --git a/examples/src/main/java/kafka/examples/Producer.java b/examples/src/main/java/kafka/examples/Producer.java index ae996a3c74907..fbfac2ef8ee82 100644 --- a/examples/src/main/java/kafka/examples/Producer.java +++ b/examples/src/main/java/kafka/examples/Producer.java @@ -35,7 +35,12 @@ public class Producer extends Thread { private int numRecords; private final CountDownLatch latch; - public Producer(String topic, Boolean isAsync, String transactionalId, int numRecords, CountDownLatch latch) { + public Producer(final String topic, + final Boolean isAsync, + final String transactionalId, + final boolean enableIdempotency, + final int numRecords, + final CountDownLatch latch) { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT); props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoProducer"); @@ -44,6 +49,8 @@ public Producer(String topic, Boolean isAsync, String transactionalId, int numRe if (transactionalId != null) { props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId); } + props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, enableIdempotency); + producer = new KafkaProducer<>(props); this.topic = topic; this.isAsync = isAsync; @@ -58,7 +65,8 @@ KafkaProducer get() { @Override public void run() { int messageNo = 0; - while (numRecords > 0) { + int recordsSent = 0; + while (recordsSent < numRecords) { String messageStr = "Message_" + messageNo; long startTime = System.currentTimeMillis(); if (isAsync) { // Send asynchronously @@ -76,8 +84,9 @@ public void run() { } } messageNo += 2; - numRecords -= 1; + recordsSent += 1; } + System.out.println("Producer sent " + numRecords + " records successfully"); latch.countDown(); } } From db292ccd153797c5f9442848eee345a95f1a10e9 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Fri, 31 Jan 2020 23:40:45 -0600 Subject: [PATCH 06/11] fix topic deletion --- .../kafka/examples/KafkaExactlyOnceDemo.java | 68 ++++++++----------- 1 file changed, 28 insertions(+), 40 deletions(-) diff --git a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java index 2bfd5701ce00d..753d6756f06e3 100644 --- a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java +++ b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java @@ -17,12 +17,8 @@ package kafka.examples; import org.apache.kafka.clients.admin.Admin; -import org.apache.kafka.clients.admin.DescribeTopicsResult; -import org.apache.kafka.clients.admin.ListTopicsResult; import org.apache.kafka.clients.admin.NewTopic; -import org.apache.kafka.clients.admin.TopicDescription; import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.common.KafkaFuture; import org.apache.kafka.common.errors.TopicExistsException; import org.apache.kafka.common.errors.UnknownTopicOrPartitionException; @@ -32,13 +28,15 @@ import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; -import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; /** * This exactly once demo driver is following below steps: - * 1. Set up a producer to pre populate a set of records into input topic - * 2. + * 1. Set up a producer to pre-populate a set of records into input topic + * 2. Set up transactional instances which does a consume-process-produce loop, tailing data from + * input topic (See {@link ExactlyOnceMessageProcessor}) + * 3. Set up a read committed consumer to verify we have all data copied from output topic, while + * the message ordering on partition level is maintained. */ public class KafkaExactlyOnceDemo { @@ -98,33 +96,23 @@ private static void recreateTopics(final int numPartitions) throws ExecutionExce List topicsToDelete = Arrays.asList(INPUT_TOPIC, OUTPUT_TOPIC); - try { - adminClient.deleteTopics(topicsToDelete).all().get(); - } catch (ExecutionException e) { - if (!(e.getCause() instanceof UnknownTopicOrPartitionException)) { - throw e; - } - System.out.println("Encountered exception during topic deletion: " + e.getCause()); - } - System.out.println("Deleted old topics: " + topicsToDelete); + deleteTopic(adminClient, topicsToDelete); // Check topic existence in a retry loop while (true) { System.out.println("Making sure the topics are deleted successfully: " + topicsToDelete); - boolean noTopicsInfo = true; Set listedTopics = adminClient.listTopics().names().get(); System.out.println("Current list of topics: " + listedTopics); - for (String topic : topicsToDelete) { - System.out.println("Checking topic " + topic); - if (listedTopics.contains(topic)) { - noTopicsInfo = false; - break; - } + boolean hasTopicInfo = false; + for (String listedTopic : listedTopics) { + if (topicsToDelete.contains(listedTopic)) { + hasTopicInfo = true; + break; + } } - - if (noTopicsInfo) { + if (!hasTopicInfo) { break; } Thread.sleep(1000); @@ -145,23 +133,23 @@ private static void recreateTopics(final int numPartitions) throws ExecutionExce throw e; } System.out.println("Metadata of the old topics are not cleared yet..."); + + deleteTopic(adminClient, topicsToDelete); + Thread.sleep(1000); } } } -} - -// DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(topicsToDelete); -// for (KafkaFuture future : describeTopicsResult.values().values()) { -// try { -// TopicDescription description = future.get(); -// System.out.println("Found topic description: " + description); -// noTopicsInfo = false; -// break; -// } catch (ExecutionException e) { -// if (!(e.getCause() instanceof UnknownTopicOrPartitionException)) { -// throw e; -// } -// } -// } + private static void deleteTopic(Admin adminClient, List topicsToDelete) throws InterruptedException, ExecutionException { + try { + adminClient.deleteTopics(topicsToDelete).all().get(); + } catch (ExecutionException e) { + if (!(e.getCause() instanceof UnknownTopicOrPartitionException)) { + throw e; + } + System.out.println("Encountered exception during topic deletion: " + e.getCause()); + } + System.out.println("Deleted old topics: " + topicsToDelete); + } +} From 392b7190313411c3949c4a707820088f3754175f Mon Sep 17 00:00:00 2001 From: abbccdda Date: Fri, 31 Jan 2020 23:45:36 -0600 Subject: [PATCH 07/11] fix group mode --- examples/README | 11 +++++++---- .../src/main/java/kafka/examples/Consumer.java | 1 - .../examples/ExactlyOnceMessageProcessor.java | 17 +++++++++-------- 3 files changed, 16 insertions(+), 13 deletions(-) diff --git a/examples/README b/examples/README index 2bdba108352e7..679e8c50649c8 100644 --- a/examples/README +++ b/examples/README @@ -6,7 +6,10 @@ To run the demo: 2. For simple consumer demo, `run bin/java-simple-consumer-demo.sh` 3. For unlimited sync-producer-consumer run, `run bin/java-producer-consumer-demo.sh sync` 4. For unlimited async-producer-consumer run, `run bin/java-producer-consumer-demo.sh` - 5. For standalone mode exactly once demo run, `run bin/exactly-once-demo.sh standaloneMode 6 3 100000`, - this means we are starting 3 EOS instances with 6 topic partitions and 100000 pre-populated records - 6. For group mode exactly once demo run, `run bin/exactly-once-demo.sh groupMode 6 3 100000`, - this means the same as the standalone demo, except consumers are using subscription mode \ No newline at end of file + 5. For standalone mode exactly once demo run, `run bin/exactly-once-demo.sh standaloneMode 6 3 50000`, + this means we are starting 3 EOS instances with 6 topic partitions and 50000 pre-populated records + 6. For group mode exactly once demo run, `run bin/exactly-once-demo.sh groupMode 6 3 50000`, + this means the same as the standalone demo, except consumers are using subscription mode + 7. Some notes for exactly once demo: + 7.1. The Kafka server has to be on broker version 2.5 or higher to be able to run group mode. + 7.2. You could also use Intellij to run the example directly by configuring parameters as "Program arguments" diff --git a/examples/src/main/java/kafka/examples/Consumer.java b/examples/src/main/java/kafka/examples/Consumer.java index df0a884539287..a35872b051746 100644 --- a/examples/src/main/java/kafka/examples/Consumer.java +++ b/examples/src/main/java/kafka/examples/Consumer.java @@ -21,7 +21,6 @@ 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.common.IsolationLevel; import java.time.Duration; import java.util.Collections; diff --git a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java index be9ae2720a223..f1fc5f82c9898 100644 --- a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java +++ b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java @@ -28,7 +28,6 @@ import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.FencedInstanceIdException; import org.apache.kafka.common.errors.ProducerFencedException; -import org.apache.kafka.common.internals.Topic; import java.time.Duration; import java.util.ArrayList; @@ -41,10 +40,8 @@ import java.util.concurrent.atomic.AtomicLong; /** - * A demo class for how to write a customized EOS app. It takes a consume-process-produce loop - * The things to pay attention to beyond a general consumer + producer app are: - * 1. Define a unique transactional.id for your app - * 2. Turn on read_committed isolation level + * A demo class for how to write a customized EOS app. It takes a consume-process-produce loop. Important + * configurations and steps are commented. */ public class ExactlyOnceMessageProcessor extends Thread { @@ -77,23 +74,25 @@ public ExactlyOnceMessageProcessor(final String mode, this.numInstances = numInstances; this.instanceIdx = instanceIdx; this.transactionalId = "Processor-" + instanceIdx; + // A unique transactional.id must be provided in order to properly use EOS. producer = new Producer(outputTopic, true, transactionalId, true, -1, null).get(); + // Consumer must be in read_committed mode, which means it won't be able to read uncommitted data. consumer = new Consumer(inputTopic, "Eos-consumer", READ_COMMITTED, -1, null).get(); this.latch = latch; } @Override public void run() { - // Init transactions call should always happen first in order to clear zombie transactions. + // Init transactions call should always happen first in order to clear zombie transactions from previous generation. producer.initTransactions(); final AtomicLong messageRemaining = new AtomicLong(Long.MAX_VALUE); // Under group mode, topic based subscription is sufficient as Consumers are safe to work transactionally after 2.5. // Under standalone mode, user needs to manually assign the topic partitions and make sure the assignment is unique - // across the consumer group. + // across the consumer group instances. if (this.mode.equals("groupMode")) { - consumer.subscribe(Collections.singleton(KafkaProperties.TOPIC), new ConsumerRebalanceListener() { + consumer.subscribe(Collections.singleton(inputTopic), new ConsumerRebalanceListener() { @Override public void onPartitionsRevoked(Collection partitions) { printWithPrefix("Revoked partition assignment to kick-off rebalancing: " + partitions); @@ -106,6 +105,7 @@ public void onPartitionsAssigned(Collection partitions) { } }); } else { + // Do a range assignment of topic partitions. List topicPartitions = new ArrayList<>(); int rangeSize = numPartitions / numInstances; int startPartition = rangeSize * instanceIdx; @@ -140,6 +140,7 @@ public void onPartitionsAssigned(Collection partitions) { producer.commitTransaction(); messageProcessed += records.count(); } catch (CommitFailedException e) { + // In case of a retriable exception, suggest aborting the ongoing transaction first for correctness. producer.abortTransaction(); } catch (ProducerFencedException | FencedInstanceIdException e) { throw new KafkaException("Encountered fatal error during processing: " + e.getMessage()); From a6752749cb3a61b01638c5d2d31e46f2f19cea3c Mon Sep 17 00:00:00 2001 From: abbccdda Date: Sat, 1 Feb 2020 09:21:43 -0800 Subject: [PATCH 08/11] style fix --- .../main/scala/kafka/controller/ControllerEventManager.scala | 2 +- core/src/main/scala/kafka/log/LogCleaner.scala | 2 +- core/src/main/scala/kafka/utils/ShutdownableThread.scala | 4 ++-- .../kafka/server/DynamicBrokerReconfigurationTest.scala | 4 ++-- .../src/test/scala/other/kafka/TestPurgatoryPerformance.scala | 2 +- examples/src/main/java/kafka/examples/Consumer.java | 2 +- 6 files changed, 8 insertions(+), 8 deletions(-) diff --git a/core/src/main/scala/kafka/controller/ControllerEventManager.scala b/core/src/main/scala/kafka/controller/ControllerEventManager.scala index eccb4a113e038..b04b85f1fa1c8 100644 --- a/core/src/main/scala/kafka/controller/ControllerEventManager.scala +++ b/core/src/main/scala/kafka/controller/ControllerEventManager.scala @@ -112,7 +112,7 @@ class ControllerEventManager(controllerId: Int, def isEmpty: Boolean = queue.isEmpty - class ControllerEventThread(name: String) extends ShutdownableThread(name = name, isInterruptable = false) { + class ControllerEventThread(name: String) extends ShutdownableThread(name = name, isInterruptible = false) { logIdent = s"[ControllerEventThread controllerId=$controllerId] " override def doWork(): Unit = { diff --git a/core/src/main/scala/kafka/log/LogCleaner.scala b/core/src/main/scala/kafka/log/LogCleaner.scala index 46c0ef876b16c..312483dfeefca 100644 --- a/core/src/main/scala/kafka/log/LogCleaner.scala +++ b/core/src/main/scala/kafka/log/LogCleaner.scala @@ -269,7 +269,7 @@ class LogCleaner(initialConfig: CleanerConfig, * choosing the dirtiest log, cleaning it, and then swapping in the cleaned segments. */ private[log] class CleanerThread(threadId: Int) - extends ShutdownableThread(name = s"kafka-log-cleaner-thread-$threadId", isInterruptable = false) { + extends ShutdownableThread(name = s"kafka-log-cleaner-thread-$threadId", isInterruptible = false) { protected override def loggerName = classOf[LogCleaner].getName diff --git a/core/src/main/scala/kafka/utils/ShutdownableThread.scala b/core/src/main/scala/kafka/utils/ShutdownableThread.scala index c16e4953d5209..0ca21c4ab166d 100644 --- a/core/src/main/scala/kafka/utils/ShutdownableThread.scala +++ b/core/src/main/scala/kafka/utils/ShutdownableThread.scala @@ -21,7 +21,7 @@ import java.util.concurrent.{CountDownLatch, TimeUnit} import org.apache.kafka.common.internals.FatalExitError -abstract class ShutdownableThread(val name: String, val isInterruptable: Boolean = true) +abstract class ShutdownableThread(val name: String, val isInterruptible: Boolean = true) extends Thread(name) with Logging { this.setDaemon(false) this.logIdent = "[" + name + "]: " @@ -50,7 +50,7 @@ abstract class ShutdownableThread(val name: String, val isInterruptable: Boolean if (isRunning) { info("Shutting down") shutdownInitiated.countDown() - if (isInterruptable) + if (isInterruptible) interrupt() true } else diff --git a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala index 36d13e72fecec..3e7fa15a15d50 100644 --- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala +++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala @@ -1613,7 +1613,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet } private class ProducerThread(clientId: String, retries: Int) - extends ShutdownableThread(clientId, isInterruptable = false) { + extends ShutdownableThread(clientId, isInterruptible = false) { private val producer = ProducerBuilder().maxRetries(retries).clientId(clientId).build() val lastSent = new ConcurrentHashMap[Int, Int]() @@ -1634,7 +1634,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet } } - private class ConsumerThread(producerThread: ProducerThread) extends ShutdownableThread("test-consumer", isInterruptable = false) { + private class ConsumerThread(producerThread: ProducerThread) extends ShutdownableThread("test-consumer", isInterruptible = false) { private val consumer = ConsumerBuilder("group1").enableAutoCommit(true).build() val lastReceived = new ConcurrentHashMap[Int, Int]() val missingRecords = new ConcurrentLinkedQueue[Int]() diff --git a/core/src/test/scala/other/kafka/TestPurgatoryPerformance.scala b/core/src/test/scala/other/kafka/TestPurgatoryPerformance.scala index 0ad5619f850dd..4b540a4f1a829 100644 --- a/core/src/test/scala/other/kafka/TestPurgatoryPerformance.scala +++ b/core/src/test/scala/other/kafka/TestPurgatoryPerformance.scala @@ -256,7 +256,7 @@ object TestPurgatoryPerformance { private class CompletionQueue { private[this] val delayQueue = new DelayQueue[Scheduled]() - private[this] val thread = new ShutdownableThread(name = "completion thread", isInterruptable = false) { + private[this] val thread = new ShutdownableThread(name = "completion thread", isInterruptible = false) { override def doWork(): Unit = { val scheduled = delayQueue.poll(100, TimeUnit.MILLISECONDS) if (scheduled != null) { diff --git a/examples/src/main/java/kafka/examples/Consumer.java b/examples/src/main/java/kafka/examples/Consumer.java index a35872b051746..43709c9141314 100644 --- a/examples/src/main/java/kafka/examples/Consumer.java +++ b/examples/src/main/java/kafka/examples/Consumer.java @@ -84,7 +84,7 @@ public String name() { } @Override - public boolean isInterruptable() { + public boolean isInterruptible() { return false; } } From 217e3db74237e5f10a7c639e52048c00401ed22d Mon Sep 17 00:00:00 2001 From: abbccdda Date: Sat, 1 Feb 2020 09:46:22 -0800 Subject: [PATCH 09/11] More comments --- examples/README | 2 +- .../main/java/kafka/examples/Consumer.java | 6 ++- .../examples/ExactlyOnceMessageProcessor.java | 44 ++++++++------- .../examples/KafkaConsumerProducerDemo.java | 2 + .../kafka/examples/KafkaExactlyOnceDemo.java | 54 +++++++++++++++---- .../main/java/kafka/examples/Producer.java | 14 ++--- 6 files changed, 83 insertions(+), 39 deletions(-) diff --git a/examples/README b/examples/README index 679e8c50649c8..2efe71ac182ee 100644 --- a/examples/README +++ b/examples/README @@ -9,7 +9,7 @@ To run the demo: 5. For standalone mode exactly once demo run, `run bin/exactly-once-demo.sh standaloneMode 6 3 50000`, this means we are starting 3 EOS instances with 6 topic partitions and 50000 pre-populated records 6. For group mode exactly once demo run, `run bin/exactly-once-demo.sh groupMode 6 3 50000`, - this means the same as the standalone demo, except consumers are using subscription mode + this means the same as the standalone demo, except consumers are using subscription mode. 7. Some notes for exactly once demo: 7.1. The Kafka server has to be on broker version 2.5 or higher to be able to run group mode. 7.2. You could also use Intellij to run the example directly by configuring parameters as "Program arguments" diff --git a/examples/src/main/java/kafka/examples/Consumer.java b/examples/src/main/java/kafka/examples/Consumer.java index 43709c9141314..19cb67ccc9158 100644 --- a/examples/src/main/java/kafka/examples/Consumer.java +++ b/examples/src/main/java/kafka/examples/Consumer.java @@ -30,6 +30,7 @@ public class Consumer extends ShutdownableThread { private final KafkaConsumer consumer; private final String topic; + private final String groupId; private final int numMessageToConsume; private int messageRemaining; private final CountDownLatch latch; @@ -40,6 +41,7 @@ public Consumer(final String topic, final int numMessageToConsume, final CountDownLatch latch) { super("KafkaConsumerExample", false); + 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); @@ -69,11 +71,11 @@ public void doWork() { consumer.subscribe(Collections.singletonList(this.topic)); ConsumerRecords records = consumer.poll(Duration.ofSeconds(1)); for (ConsumerRecord record : records) { - System.out.println("Verify-consumer received message : from partition " + record.partition() + ", (" + record.key() + ", " + record.value() + ") at offset " + record.offset()); + System.out.println(groupId + " received message : from partition " + record.partition() + ", (" + record.key() + ", " + record.value() + ") at offset " + record.offset()); } messageRemaining -= records.count(); if (messageRemaining <= 0) { - System.out.println("Verify-consumer finished reading " + numMessageToConsume + " messages"); + System.out.println(groupId + " finished reading " + numMessageToConsume + " messages"); latch.countDown(); } } diff --git a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java index f1fc5f82c9898..e37d3dcb9db65 100644 --- a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java +++ b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java @@ -40,8 +40,8 @@ import java.util.concurrent.atomic.AtomicLong; /** - * A demo class for how to write a customized EOS app. It takes a consume-process-produce loop. Important - * configurations and steps are commented. + * A demo class for how to write a customized EOS app. It takes a consume-process-produce loop. + * Important configurations and APIs are commented. */ public class ExactlyOnceMessageProcessor extends Thread { @@ -50,6 +50,7 @@ public class ExactlyOnceMessageProcessor extends Thread { private final String mode; private final String inputTopic; private final String outputTopic; + private final String consumerGroupId; private final int numPartitions; private final int numInstances; private final int instanceIdx; @@ -70,6 +71,7 @@ public ExactlyOnceMessageProcessor(final String mode, this.mode = mode; this.inputTopic = inputTopic; this.outputTopic = outputTopic; + this.consumerGroupId = "Eos-consumer"; this.numPartitions = numPartitions; this.numInstances = numInstances; this.instanceIdx = instanceIdx; @@ -77,7 +79,7 @@ public ExactlyOnceMessageProcessor(final String mode, // A unique transactional.id must be provided in order to properly use EOS. producer = new Producer(outputTopic, true, transactionalId, true, -1, null).get(); // Consumer must be in read_committed mode, which means it won't be able to read uncommitted data. - consumer = new Consumer(inputTopic, "Eos-consumer", READ_COMMITTED, -1, null).get(); + consumer = new Consumer(inputTopic, consumerGroupId, READ_COMMITTED, -1, null).get(); this.latch = latch; } @@ -88,19 +90,19 @@ public void run() { final AtomicLong messageRemaining = new AtomicLong(Long.MAX_VALUE); - // Under group mode, topic based subscription is sufficient as Consumers are safe to work transactionally after 2.5. + // Under group mode, topic based subscription is sufficient as EOS apps are safe to cooperate transactionally after 2.5. // Under standalone mode, user needs to manually assign the topic partitions and make sure the assignment is unique // across the consumer group instances. if (this.mode.equals("groupMode")) { consumer.subscribe(Collections.singleton(inputTopic), new ConsumerRebalanceListener() { @Override public void onPartitionsRevoked(Collection partitions) { - printWithPrefix("Revoked partition assignment to kick-off rebalancing: " + partitions); + printWithTxnId("Revoked partition assignment to kick-off rebalancing: " + partitions); } @Override public void onPartitionsAssigned(Collection partitions) { - printWithPrefix("Received partition assignment after rebalancing: " + partitions); + printWithTxnId("Received partition assignment after rebalancing: " + partitions); messageRemaining.set(messagesRemaining(consumer)); } }); @@ -115,7 +117,7 @@ public void onPartitionsAssigned(Collection partitions) { } consumer.assign(topicPartitions); - printWithPrefix("Manually assign partitions: " + topicPartitions); + printWithTxnId("Manually assign partitions: " + topicPartitions); } int messageProcessed = 0; @@ -126,8 +128,8 @@ public void onPartitionsAssigned(Collection partitions) { // Begin a new transaction session. producer.beginTransaction(); for (ConsumerRecord record : records) { + // Process the record and send to downstream. ProducerRecord customizedRecord = transform(record); - // Send records to the downstream. producer.send(customizedRecord); } Map positions = new HashMap<>(); @@ -135,8 +137,14 @@ public void onPartitionsAssigned(Collection partitions) { positions.put(topicPartition, new OffsetAndMetadata(consumer.position(topicPartition), null)); } // Checkpoint the progress by sending offsets to group coordinator broker. - producer.sendOffsetsToTransaction(positions, consumer.groupMetadata()); - // Finish the transaction. + // Under group mode, we must apply consumer group metadata for proper fencing. + if (this.mode.equals("groupMode")) { + producer.sendOffsetsToTransaction(positions, consumer.groupMetadata()); + } else { + producer.sendOffsetsToTransaction(positions, consumerGroupId); + } + + // Finish the transaction. All sent records should be visible for consumption now. producer.commitTransaction(); messageProcessed += records.count(); } catch (CommitFailedException e) { @@ -147,32 +155,32 @@ public void onPartitionsAssigned(Collection partitions) { } } messageRemaining.set(messagesRemaining(consumer)); - printWithPrefix("Message remaining: " + messageRemaining); + printWithTxnId("Message remaining: " + messageRemaining); } - printWithPrefix("Finished processing " + messageProcessed + " records"); + printWithTxnId("Finished processing " + messageProcessed + " records"); latch.countDown(); } - private void printWithPrefix(String message) { + private void printWithTxnId(final String message) { System.out.println(transactionalId + ": " + message); } - private ProducerRecord transform(ConsumerRecord record) { - printWithPrefix("Transformed record (" + record.key() + "," + record.value() + ")"); + private ProducerRecord transform(final ConsumerRecord record) { + printWithTxnId("Transformed record (" + record.key() + "," + record.value() + ")"); return new ProducerRecord<>(outputTopic, record.key() / 2, "Transformed_" + record.value()); } - private long messagesRemaining(KafkaConsumer consumer) { - // If we couldn't detect any end offset, that means we are still not able to fetch offsets. + private long messagesRemaining(final KafkaConsumer consumer) { final Map fullEndOffsets = consumer.endOffsets(new ArrayList<>(consumer.assignment())); + // If we couldn't detect any end offset, that means we are still not able to fetch offsets. if (fullEndOffsets.isEmpty()) { return Long.MAX_VALUE; } return consumer.assignment().stream().mapToLong(partition -> { long currentPosition = consumer.position(partition); - printWithPrefix("Processing partition " + partition + " with full offsets " + fullEndOffsets); + printWithTxnId("Processing partition " + partition + " with full offsets " + fullEndOffsets); if (fullEndOffsets.containsKey(partition)) { return fullEndOffsets.get(partition) - currentPosition; } diff --git a/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java b/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java index 9a4a50923971c..561732bec6c1a 100644 --- a/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java +++ b/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java @@ -29,5 +29,7 @@ public static void main(String[] args) throws InterruptedException { Consumer consumerThread = new Consumer(KafkaProperties.TOPIC, "DemoConsumer", false, 10000, latch); consumerThread.start(); latch.await(5, TimeUnit.MINUTES); + consumerThread.shutdown(); + System.out.println("All finished!"); } } diff --git a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java index 753d6756f06e3..10c6100971a7c 100644 --- a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java +++ b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java @@ -31,12 +31,39 @@ import java.util.concurrent.TimeUnit; /** - * This exactly once demo driver is following below steps: - * 1. Set up a producer to pre-populate a set of records into input topic - * 2. Set up transactional instances which does a consume-process-produce loop, tailing data from - * input topic (See {@link ExactlyOnceMessageProcessor}) - * 3. Set up a read committed consumer to verify we have all data copied from output topic, while - * the message ordering on partition level is maintained. + * This exactly once demo driver takes 4 arguments: + * - mode: whether to run as standalone app, or a group + * - partition: number of partitions for input/output topic + * - instances: number of instances + * - records: number of records + * An example argument list would be `groupMode 6 3 50000` + * + * The driver could be decomposed as following stages: + * + * 1. Cleanup any topic whose name conflicts with input and output topic, so that we have a clean-start. + * + * 2. Set up a producer in a separate thread to pre-populate a set of records with even number keys into + * the input topic. The driver will block for the record generation to finish, so the producer + * must be in synchronous sending mode. + * + * 3. Set up transactional instances in separate threads which does a consume-process-produce loop, + * tailing data from input topic (See {@link ExactlyOnceMessageProcessor}). Each EOS instance will + * drain all the records from either given partitions or auto assigned partitions by actively + * comparing log end offset with committed offset. Each record will be processed exactly once + * as dividing the key by 2, and extend the value message. The driver will block for all the record + * processing to finish. The transformed record shall be written to the output topic, with + * transactional guarantee. + * + * 4. Set up a read committed consumer in a separate thread to verify we have all records within + * the output topic, while the message ordering on partition level is maintained. + * The driver will block for the consumption of all committed records. + * + * From this demo, you could see that all the records from pre-population are processed exactly once, + * in either standalone mode or group mode, with strong partition level ordering guarantee. + * + * Note: please start the kafka broker and zookeeper in local first. The broker version must be >= 2.5 + * in order to run group mode, otherwise the app could throw + * {@link org.apache.kafka.common.errors.UnsupportedVersionException}. */ public class KafkaExactlyOnceDemo { @@ -54,11 +81,12 @@ public static void main(String[] args) throws InterruptedException, ExecutionExc int numInstances = Integer.valueOf(args[2]); int numRecords = Integer.valueOf(args[3]); + /* Stage 1: topic cleanup and recreation */ recreateTopics(numPartitions); CountDownLatch prePopulateLatch = new CountDownLatch(1); - // Pre-populate records. + /* Stage 2: pre-populate records */ final boolean isAsync = false; final boolean enableIdempotency = true; Producer producerThread = new Producer(INPUT_TOPIC, isAsync, null, enableIdempotency, numRecords, prePopulateLatch); @@ -68,7 +96,7 @@ public static void main(String[] args) throws InterruptedException, ExecutionExc CountDownLatch transactionalCopyLatch = new CountDownLatch(numInstances); - // Transactionally copy over all messages. + /* Stage 3: transactionally process all messages */ for (int instanceIdx = 0; instanceIdx < numInstances; instanceIdx++) { ExactlyOnceMessageProcessor messageProcessor = new ExactlyOnceMessageProcessor(mode, INPUT_TOPIC, OUTPUT_TOPIC, numPartitions, @@ -80,6 +108,7 @@ public static void main(String[] args) throws InterruptedException, ExecutionExc CountDownLatch consumeLatch = new CountDownLatch(1); + /* Stage 4: consume all processed messages to verify exactly once */ Consumer consumerThread = new Consumer(OUTPUT_TOPIC, "Verify-consumer", true, numRecords, consumeLatch); consumerThread.start(); @@ -88,9 +117,11 @@ public static void main(String[] args) throws InterruptedException, ExecutionExc System.out.println("All finished!"); } - private static void recreateTopics(final int numPartitions) throws ExecutionException, InterruptedException { + private static void recreateTopics(final int numPartitions) + throws ExecutionException, InterruptedException { Properties props = new Properties(); - props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, + KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT); Admin adminClient = Admin.create(props); @@ -141,7 +172,8 @@ private static void recreateTopics(final int numPartitions) throws ExecutionExce } } - private static void deleteTopic(Admin adminClient, List topicsToDelete) throws InterruptedException, ExecutionException { + private static void deleteTopic(final Admin adminClient, final List topicsToDelete) + throws InterruptedException, ExecutionException { try { adminClient.deleteTopics(topicsToDelete).all().get(); } catch (ExecutionException e) { diff --git a/examples/src/main/java/kafka/examples/Producer.java b/examples/src/main/java/kafka/examples/Producer.java index fbfac2ef8ee82..3805dd389a4f5 100644 --- a/examples/src/main/java/kafka/examples/Producer.java +++ b/examples/src/main/java/kafka/examples/Producer.java @@ -64,26 +64,26 @@ KafkaProducer get() { @Override public void run() { - int messageNo = 0; + int messageKey = 0; int recordsSent = 0; while (recordsSent < numRecords) { - String messageStr = "Message_" + messageNo; + String messageStr = "Message_" + messageKey; long startTime = System.currentTimeMillis(); if (isAsync) { // Send asynchronously producer.send(new ProducerRecord<>(topic, - messageNo, - messageStr), new DemoCallBack(startTime, messageNo, messageStr)); + messageKey, + messageStr), new DemoCallBack(startTime, messageKey, messageStr)); } else { // Send synchronously try { producer.send(new ProducerRecord<>(topic, - messageNo, + messageKey, messageStr)).get(); - System.out.println("Sent message: (" + messageNo + ", " + messageStr + ")"); + System.out.println("Sent message: (" + messageKey + ", " + messageStr + ")"); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } } - messageNo += 2; + messageKey += 2; recordsSent += 1; } System.out.println("Producer sent " + numRecords + " records successfully"); From 5827af7522abb6206e61733180e39200811338c3 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Tue, 4 Feb 2020 15:37:33 -0800 Subject: [PATCH 10/11] guozhang's comment --- .../src/main/java/kafka/examples/KafkaExactlyOnceDemo.java | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java index 10c6100971a7c..d418ebaa88541 100644 --- a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java +++ b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java @@ -87,9 +87,7 @@ public static void main(String[] args) throws InterruptedException, ExecutionExc CountDownLatch prePopulateLatch = new CountDownLatch(1); /* Stage 2: pre-populate records */ - final boolean isAsync = false; - final boolean enableIdempotency = true; - Producer producerThread = new Producer(INPUT_TOPIC, isAsync, null, enableIdempotency, numRecords, prePopulateLatch); + Producer producerThread = new Producer(INPUT_TOPIC, false, null, true, numRecords, prePopulateLatch); producerThread.start(); prePopulateLatch.await(5, TimeUnit.MINUTES); From 7a748240c100014e86e831f614bf4219974fbd32 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Wed, 5 Feb 2020 12:44:15 -0800 Subject: [PATCH 11/11] move abort txn out and add commit reset --- .../examples/ExactlyOnceMessageProcessor.java | 23 +++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java index e37d3dcb9db65..53685f351503d 100644 --- a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java +++ b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java @@ -121,10 +121,18 @@ public void onPartitionsAssigned(Collection partitions) { } int messageProcessed = 0; + boolean abortPreviousTransaction = false; while (messageRemaining.get() > 0) { ConsumerRecords records = consumer.poll(Duration.ofMillis(200)); if (records.count() > 0) { try { + // Abort previous transaction if instructed. + if (abortPreviousTransaction) { + producer.abortTransaction(); + // The consumer fetch position also needs to be reset. + resetToLastCommittedPositions(consumer); + abortPreviousTransaction = false; + } // Begin a new transaction session. producer.beginTransaction(); for (ConsumerRecord record : records) { @@ -148,8 +156,8 @@ public void onPartitionsAssigned(Collection partitions) { producer.commitTransaction(); messageProcessed += records.count(); } catch (CommitFailedException e) { - // In case of a retriable exception, suggest aborting the ongoing transaction first for correctness. - producer.abortTransaction(); + // In case of a retriable exception, suggest aborting the ongoing transaction for correctness. + abortPreviousTransaction = true; } catch (ProducerFencedException | FencedInstanceIdException e) { throw new KafkaException("Encountered fatal error during processing: " + e.getMessage()); } @@ -187,4 +195,15 @@ private long messagesRemaining(final KafkaConsumer consumer) { return 0; }).sum(); } + + private static void resetToLastCommittedPositions(KafkaConsumer consumer) { + final Map committed = consumer.committed(consumer.assignment()); + consumer.assignment().forEach(tp -> { + OffsetAndMetadata offsetAndMetadata = committed.get(tp); + if (offsetAndMetadata != null) + consumer.seek(tp, offsetAndMetadata.offset()); + else + consumer.seekToBeginning(Collections.singleton(tp)); + }); + } }