root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Add timeout for closing producers.

This commit is contained in:
Ewen Cheslack-Postava 2016-08-12 14:00:53 -07:00
parent 98ec7f69bb
commit 7d3972479b
2 changed files with 8 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
View File

@ -115,7 +115,7 @@ class WorkerSourceTask extends WorkerTask {
}
protected void close() {
producer.close();
producer.close(30, TimeUnit.SECONDS);
}
@Override

14
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
View File

@ -156,7 +156,7 @@ public class WorkerSourceTaskTest extends ThreadedTest {
statusListener.onShutdown(taskId);
EasyMock.expectLastCall();
producer.close();
producer.close(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class));
EasyMock.expectLastCall();
PowerMock.replayAll();
@ -198,7 +198,7 @@ public class WorkerSourceTaskTest extends ThreadedTest {
statusListener.onShutdown(taskId);
EasyMock.expectLastCall();
producer.close();
producer.close(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class));
EasyMock.expectLastCall();
PowerMock.replayAll();
@ -244,7 +244,7 @@ public class WorkerSourceTaskTest extends ThreadedTest {
statusListener.onShutdown(taskId);
EasyMock.expectLastCall();
producer.close();
producer.close(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class));
EasyMock.expectLastCall();
PowerMock.replayAll();
@ -289,7 +289,7 @@ public class WorkerSourceTaskTest extends ThreadedTest {
EasyMock.expectLastCall();
expectOffsetFlush(true);
producer.close();
producer.close(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class));
EasyMock.expectLastCall();
PowerMock.replayAll();
@ -329,7 +329,7 @@ public class WorkerSourceTaskTest extends ThreadedTest {
statusListener.onShutdown(taskId);
EasyMock.expectLastCall();
producer.close();
producer.close(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class));
EasyMock.expectLastCall();
PowerMock.replayAll();
@ -370,7 +370,7 @@ public class WorkerSourceTaskTest extends ThreadedTest {
statusListener.onShutdown(taskId);
EasyMock.expectLastCall();
producer.close();
producer.close(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class));
EasyMock.expectLastCall();
PowerMock.replayAll();
@ -516,7 +516,7 @@ public class WorkerSourceTaskTest extends ThreadedTest {
statusListener.onShutdown(taskId);
EasyMock.expectLastCall();
producer.close();
producer.close(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class));
EasyMock.expectLastCall();
PowerMock.replayAll();