Repository: kafka Updated Branches: refs/heads/trunk 5b4e299a4 -> 254e3b77d
KAFKA-4699; Invoke producer callbacks before completing the future This behaviour was changed in 8b3c6c0, but it caused interceptor test failures (which rely on callbacks) and since weâre so close to code freeze, itâs better to be conservative. Author: Ismael Juma <[email protected]> Reviewers: Jason Gustafson <[email protected]> Closes #2440 from ijuma/kafka-4699-callbacks-invoked-before-future-is-completed Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/254e3b77 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/254e3b77 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/254e3b77 Branch: refs/heads/trunk Commit: 254e3b77d656a610f19efd1124802e073dfda4b8 Parents: 5b4e299 Author: Ismael Juma <[email protected]> Authored: Thu Jan 26 09:50:25 2017 -0800 Committer: Jason Gustafson <[email protected]> Committed: Thu Jan 26 09:50:25 2017 -0800 ---------------------------------------------------------------------- .../kafka/clients/producer/MockProducer.java | 3 ++- .../internals/ProduceRequestResult.java | 14 +++++++++--- .../clients/producer/internals/RecordBatch.java | 5 ++-- .../kafka/clients/producer/RecordSendTest.java | 6 +++-- .../runtime/SourceTaskOffsetCommitterTest.java | 2 +- .../kafka/api/PlaintextConsumerTest.scala | 24 ++++++++++---------- 6 files changed, 33 insertions(+), 21 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/254e3b77/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java index 008ca70..165437d 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java @@ -245,13 +245,14 @@ public class MockProducer<K, V> implements Producer<K, V> { } public void complete(RuntimeException e) { - result.done(e == null ? offset : -1L, Record.NO_TIMESTAMP, e); + result.set(e == null ? offset : -1L, Record.NO_TIMESTAMP, e); if (callback != null) { if (e == null) callback.onCompletion(metadata, null); else callback.onCompletion(null, e); } + result.done(); } } http://git-wip-us.apache.org/repos/asf/kafka/blob/254e3b77/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProduceRequestResult.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProduceRequestResult.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProduceRequestResult.java index 41c204d..a98b28e 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProduceRequestResult.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProduceRequestResult.java @@ -34,7 +34,7 @@ public final class ProduceRequestResult { private final CountDownLatch latch = new CountDownLatch(1); private final TopicPartition topicPartition; - private volatile long baseOffset = -1L; + private volatile Long baseOffset = null; private volatile long logAppendTime = Record.NO_TIMESTAMP; private volatile RuntimeException error; @@ -48,16 +48,24 @@ public final class ProduceRequestResult { } /** - * Mark this request as complete and unblock any threads waiting on its completion. + * Set the result of the produce request. * * @param baseOffset The base offset assigned to the record * @param logAppendTime The log append time or -1 if CreateTime is being used * @param error The error that occurred if there was one, or null */ - public void done(long baseOffset, long logAppendTime, RuntimeException error) { + public void set(long baseOffset, long logAppendTime, RuntimeException error) { this.baseOffset = baseOffset; this.logAppendTime = logAppendTime; this.error = error; + } + + /** + * Mark this request as complete and unblock any threads waiting on its completion. + */ + public void done() { + if (baseOffset == null) + throw new IllegalStateException("The method `set` must be invoked before this method."); this.latch.countDown(); } http://git-wip-us.apache.org/repos/asf/kafka/blob/254e3b77/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java index 1b751bf..c8eddd5 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java @@ -93,8 +93,8 @@ public final class RecordBatch { log.trace("Produced messages to topic-partition {} with base offset offset {} and error: {}.", topicPartition, baseOffset, exception); - // Complete the future before invoking the callbacks as we rely on its state for the `onCompletion` call - produceFuture.done(baseOffset, logAppendTime, exception); + // Set the future before invoking the callbacks as we rely on its state for the `onCompletion` call + produceFuture.set(baseOffset, logAppendTime, exception); // execute callbacks for (Thunk thunk : thunks) { @@ -110,6 +110,7 @@ public final class RecordBatch { } } + produceFuture.done(); } /** http://git-wip-us.apache.org/repos/asf/kafka/blob/254e3b77/clients/src/test/java/org/apache/kafka/clients/producer/RecordSendTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/RecordSendTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/RecordSendTest.java index a420d61..bc8105b 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/RecordSendTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/RecordSendTest.java @@ -54,7 +54,8 @@ public class RecordSendTest { } catch (TimeoutException e) { /* this is good */ } - request.done(baseOffset, Record.NO_TIMESTAMP, null); + request.set(baseOffset, Record.NO_TIMESTAMP, null); + request.done(); assertTrue(future.isDone()); assertEquals(baseOffset + relOffset, future.get().offset()); } @@ -86,7 +87,8 @@ public class RecordSendTest { public void run() { try { sleep(timeout); - request.done(baseOffset, Record.NO_TIMESTAMP, error); + request.set(baseOffset, Record.NO_TIMESTAMP, error); + request.done(); } catch (InterruptedException e) { } } }; http://git-wip-us.apache.org/repos/asf/kafka/blob/254e3b77/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitterTest.java ---------------------------------------------------------------------- diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitterTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitterTest.java index 45125cc..d0fa1c8 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitterTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitterTest.java @@ -165,7 +165,7 @@ public class SourceTaskOffsetCommitterTest extends ThreadedTest { EasyMock.expect(task.cancel(eq(false))).andReturn(false); EasyMock.expect(task.isDone()).andReturn(false); EasyMock.expect(task.get()).andThrow(new CancellationException()); - mockLog.trace(EasyMock.anyString(), EasyMock.anyObject()); + mockLog.trace(EasyMock.anyString(), EasyMock.<Object>anyObject()); PowerMock.expectLastCall(); PowerMock.replayAll(); http://git-wip-us.apache.org/repos/asf/kafka/blob/254e3b77/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala index 4fa1462..50941ce 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala @@ -847,28 +847,28 @@ class PlaintextConsumerTest extends BaseConsumerTest { producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList) producerProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "org.apache.kafka.test.MockProducerInterceptor") producerProps.put("mock.interceptor.append", appendStr) - val testProducer = new KafkaProducer[String, String](producerProps, new StringSerializer, new StringSerializer) + val testProducer = new KafkaProducer(producerProps, new StringSerializer, new StringSerializer) // produce records val numRecords = 10 (0 until numRecords).map { i => - testProducer.send(new ProducerRecord(tp.topic(), tp.partition(), s"key $i", s"value $i")) + testProducer.send(new ProducerRecord(tp.topic, tp.partition, s"key $i", s"value $i")) }.foreach(_.get) - assertEquals(numRecords, MockProducerInterceptor.ONSEND_COUNT.intValue()) - assertEquals(numRecords, MockProducerInterceptor.ON_SUCCESS_COUNT.intValue()) + assertEquals(numRecords, MockProducerInterceptor.ONSEND_COUNT.intValue) + assertEquals(numRecords, MockProducerInterceptor.ON_SUCCESS_COUNT.intValue) // send invalid record try { - testProducer.send(null, null) + testProducer.send(null) fail("Should not allow sending a null record") } catch { case _: Throwable => - assertEquals("Interceptor should be notified about exception", 1, MockProducerInterceptor.ON_ERROR_COUNT.intValue()) + assertEquals("Interceptor should be notified about exception", 1, MockProducerInterceptor.ON_ERROR_COUNT.intValue) assertEquals("Interceptor should not receive metadata with an exception when record is null", 0, MockProducerInterceptor.ON_ERROR_WITH_METADATA_COUNT.intValue()) } // create consumer with interceptor this.consumerConfig.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, "org.apache.kafka.test.MockConsumerInterceptor") - val testConsumer = new KafkaConsumer[String, String](this.consumerConfig, new StringDeserializer(), new StringDeserializer()) + val testConsumer = new KafkaConsumer(this.consumerConfig, new StringDeserializer, new StringDeserializer) testConsumer.assign(List(tp).asJava) testConsumer.seek(tp, 0) @@ -876,22 +876,22 @@ class PlaintextConsumerTest extends BaseConsumerTest { val records = consumeRecords(testConsumer, numRecords) for (i <- 0 until numRecords) { val record = records(i) - assertEquals(s"key $i", new String(record.key())) - assertEquals(s"value $i$appendStr".toUpperCase(Locale.ROOT), new String(record.value())) + assertEquals(s"key $i", new String(record.key)) + assertEquals(s"value $i$appendStr".toUpperCase(Locale.ROOT), new String(record.value)) } // commit sync and verify onCommit is called - val commitCountBefore = MockConsumerInterceptor.ON_COMMIT_COUNT.intValue() + val commitCountBefore = MockConsumerInterceptor.ON_COMMIT_COUNT.intValue testConsumer.commitSync(Map[TopicPartition, OffsetAndMetadata]((tp, new OffsetAndMetadata(2L))).asJava) assertEquals(2, testConsumer.committed(tp).offset) - assertEquals(commitCountBefore + 1, MockConsumerInterceptor.ON_COMMIT_COUNT.intValue()) + assertEquals(commitCountBefore + 1, MockConsumerInterceptor.ON_COMMIT_COUNT.intValue) // commit async and verify onCommit is called val commitCallback = new CountConsumerCommitCallback() testConsumer.commitAsync(Map[TopicPartition, OffsetAndMetadata]((tp, new OffsetAndMetadata(5L))).asJava, commitCallback) awaitCommitCallback(testConsumer, commitCallback) assertEquals(5, testConsumer.committed(tp).offset) - assertEquals(commitCountBefore + 2, MockConsumerInterceptor.ON_COMMIT_COUNT.intValue()) + assertEquals(commitCountBefore + 2, MockConsumerInterceptor.ON_COMMIT_COUNT.intValue) testConsumer.close() testProducer.close()
