You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by rh...@apache.org on 2019/10/16 01:14:30 UTC
[kafka] branch 2.4 updated: KAFKA-9014: Fix AssertionError when
SourceTask.poll returns an empty list (#7491)
This is an automated email from the ASF dual-hosted git repository.
rhauch pushed a commit to branch 2.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/2.4 by this push:
new 5a8ad0a KAFKA-9014: Fix AssertionError when SourceTask.poll returns an empty list (#7491)
5a8ad0a is described below
commit 5a8ad0a8024bcc7e9111fbd74dd35e02bea0491b
Author: Konstantine Karantasis <ko...@confluent.io>
AuthorDate: Tue Oct 15 14:08:31 2019 -0700
KAFKA-9014: Fix AssertionError when SourceTask.poll returns an empty list (#7491)
Author: Konstantine Karantasis <ko...@confluent.io>
Reviewer: Randall Hauch <rh...@gmail.com>
---
.../kafka/connect/runtime/WorkerSourceTask.java | 3 +-
.../connect/runtime/WorkerSourceTaskTest.java | 64 ++++++++++++++++++++++
2 files changed, 66 insertions(+), 1 deletion(-)
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
index c66429d..5e24d41 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
@@ -300,7 +300,8 @@ class WorkerSourceTask extends WorkerTask {
private boolean sendRecords() {
int processed = 0;
recordBatch(toSend.size());
- final SourceRecordWriteCounter counter = new SourceRecordWriteCounter(toSend.size(), sourceTaskMetricsGroup);
+ final SourceRecordWriteCounter counter =
+ toSend.size() > 0 ? new SourceRecordWriteCounter(toSend.size(), sourceTaskMetricsGroup) : null;
for (final SourceRecord preTransformRecord : toSend) {
maybeThrowProducerSendException();
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
index dff267a..e803931 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
@@ -84,6 +84,7 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@PowerMockIgnore({"javax.management.*",
+ "org.apache.log4j.*",
"org.apache.kafka.connect.runtime.isolation.*"})
@RunWith(PowerMockRunner.class)
public class WorkerSourceTaskTest extends ThreadedTest {
@@ -351,6 +352,51 @@ public class WorkerSourceTaskTest extends ThreadedTest {
}
@Test
+ public void testPollReturnsNoRecords() throws Exception {
+ // Test that the task handles an empty list of records
+ createWorkerTask();
+
+ sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
+ EasyMock.expectLastCall();
+ sourceTask.start(TASK_PROPS);
+ EasyMock.expectLastCall();
+ statusListener.onStartup(taskId);
+ EasyMock.expectLastCall();
+
+ // We'll wait for some data, then trigger a flush
+ final CountDownLatch pollLatch = expectEmptyPolls(1, new AtomicInteger());
+ expectOffsetFlush(true);
+
+ sourceTask.stop();
+ EasyMock.expectLastCall();
+ expectOffsetFlush(true);
+
+ statusListener.onShutdown(taskId);
+ EasyMock.expectLastCall();
+
+ producer.close(EasyMock.anyObject(Duration.class));
+ EasyMock.expectLastCall();
+
+ transformationChain.close();
+ EasyMock.expectLastCall();
+
+ PowerMock.replayAll();
+
+ workerTask.initialize(TASK_CONFIG);
+ Future<?> taskFuture = executor.submit(workerTask);
+
+ assertTrue(awaitLatch(pollLatch));
+ assertTrue(workerTask.commitOffsets());
+ workerTask.stop();
+ assertTrue(workerTask.awaitStop(1000));
+
+ taskFuture.get();
+ assertPollMetrics(0);
+
+ PowerMock.verifyAll();
+ }
+
+ @Test
public void testCommit() throws Exception {
// Test that the task commits properly when prompted
createWorkerTask();
@@ -767,6 +813,24 @@ public class WorkerSourceTaskTest extends ThreadedTest {
PowerMock.verifyAll();
}
+ private CountDownLatch expectEmptyPolls(int minimum, final AtomicInteger count) throws InterruptedException {
+ final CountDownLatch latch = new CountDownLatch(minimum);
+ // Note that we stub these to allow any number of calls because the thread will continue to
+ // run. The count passed in + latch returned just makes sure we get *at least* that number of
+ // calls
+ EasyMock.expect(sourceTask.poll())
+ .andStubAnswer(new IAnswer<List<SourceRecord>>() {
+ @Override
+ public List<SourceRecord> answer() throws Throwable {
+ count.incrementAndGet();
+ latch.countDown();
+ Thread.sleep(10);
+ return Collections.emptyList();
+ }
+ });
+ return latch;
+ }
+
private CountDownLatch expectPolls(int minimum, final AtomicInteger count) throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(minimum);
// Note that we stub these to allow any number of calls because the thread will continue to