You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ce...@apache.org on 2023/07/10 16:59:00 UTC
[kafka] branch trunk updated: KAFKA-14059 Replace PowerMock with Mockito in WorkerSourceTaskTest (#13383)
This is an automated email from the ASF dual-hosted git repository.
cegerton pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 6368d14a1d8 KAFKA-14059 Replace PowerMock with Mockito in WorkerSourceTaskTest (#13383)
6368d14a1d8 is described below
commit 6368d14a1d8c37305290b8b89fb5990ad07aa4db
Author: Hector Geraldino <hg...@gmail.com>
AuthorDate: Mon Jul 10 12:58:54 2023 -0400
KAFKA-14059 Replace PowerMock with Mockito in WorkerSourceTaskTest (#13383)
Reviewers: Chris Egerton <ch...@aiven.io>
---
build.gradle | 3 +-
.../kafka/connect/runtime/WorkerSourceTask.java | 6 +-
.../connect/runtime/WorkerSourceTaskTest.java | 850 +++++++++------------
3 files changed, 366 insertions(+), 493 deletions(-)
diff --git a/build.gradle b/build.gradle
index b47b850625c..eef7aa68abd 100644
--- a/build.gradle
+++ b/build.gradle
@@ -419,8 +419,7 @@ subprojects {
"**/DistributedHerderTest.*",
"**/KafkaConfigBackingStoreTest.*",
"**/KafkaBasedLogTest.*", "**/StandaloneHerderTest.*",
- "**/WorkerSinkTaskTest.*", "**/WorkerSinkTaskThreadedTest.*",
- "**/WorkerSourceTaskTest.*"
+ "**/WorkerSinkTaskTest.*", "**/WorkerSinkTaskThreadedTest.*"
])
}
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 a6767675a3a..501018cdce5 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
@@ -58,7 +58,8 @@ class WorkerSourceTask extends AbstractWorkerSourceTask {
private static final Logger log = LoggerFactory.getLogger(WorkerSourceTask.class);
private volatile CommittableOffsets committableOffsets;
- private final SubmittedRecords submittedRecords;
+ //VisibleForTesting
+ final SubmittedRecords submittedRecords;
private final AtomicReference<Exception> producerSendException;
public WorkerSourceTask(ConnectorTaskId id,
@@ -320,7 +321,8 @@ class WorkerSourceTask extends AbstractWorkerSourceTask {
return true;
}
- private void updateCommittableOffsets() {
+ // Visible for testing
+ void updateCommittableOffsets() {
CommittableOffsets newOffsets = submittedRecords.committableOffsets();
synchronized (this) {
this.committableOffsets = this.committableOffsets.updatedWith(newOffsets);
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 928f5d89430..ec57aea3067 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
@@ -17,14 +17,14 @@
package org.apache.kafka.connect.runtime;
import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.InvalidTopicException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
-import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.utils.Time;
@@ -48,34 +48,33 @@ import org.apache.kafka.connect.storage.HeaderConverter;
import org.apache.kafka.connect.storage.OffsetStorageWriter;
import org.apache.kafka.connect.storage.StatusBackingStore;
import org.apache.kafka.connect.storage.StringConverter;
-import org.apache.kafka.connect.util.Callback;
+import org.apache.kafka.connect.test.util.ConcurrencyUtils;
import org.apache.kafka.connect.util.ConnectorTaskId;
import org.apache.kafka.connect.util.ParameterizedTest;
import org.apache.kafka.connect.util.TopicAdmin;
import org.apache.kafka.connect.util.TopicCreationGroup;
import org.apache.kafka.common.utils.LogCaptureAppender;
-import org.easymock.Capture;
-import org.easymock.EasyMock;
-import org.easymock.IAnswer;
-import org.easymock.IExpectationSetters;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
-import org.powermock.api.easymock.PowerMock;
-import org.powermock.api.easymock.annotation.Mock;
-import org.powermock.api.easymock.annotation.MockStrict;
-import org.powermock.core.classloader.annotations.PowerMockIgnore;
-import org.powermock.modules.junit4.PowerMockRunner;
-import org.powermock.modules.junit4.PowerMockRunnerDelegate;
-import org.powermock.reflect.Whitebox;
+import org.junit.runners.Parameterized;
+import org.mockito.AdditionalAnswers;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnit;
+import org.mockito.junit.MockitoRule;
+import org.mockito.quality.Strictness;
+import org.mockito.stubbing.Answer;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -86,6 +85,7 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
import static org.apache.kafka.connect.integration.MonitorableSourceConnector.TOPIC_CONFIG;
import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLASS_CONFIG;
@@ -99,19 +99,36 @@ import static org.apache.kafka.connect.runtime.TopicCreationConfig.INCLUDE_REGEX
import static org.apache.kafka.connect.runtime.TopicCreationConfig.PARTITIONS_CONFIG;
import static org.apache.kafka.connect.runtime.TopicCreationConfig.REPLICATION_FACTOR_CONFIG;
import static org.apache.kafka.connect.runtime.WorkerConfig.TOPIC_CREATION_ENABLE_CONFIG;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
-
-@PowerMockIgnore({"javax.management.*",
- "org.apache.log4j.*"})
-@RunWith(PowerMockRunner.class)
-@PowerMockRunnerDelegate(ParameterizedTest.class)
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+@SuppressWarnings({"unchecked"})
+@RunWith(Parameterized.class)
public class WorkerSourceTaskTest {
+
+ public static final String POLL_TIMEOUT_MSG = "Timeout waiting for poll";
+ @org.junit.Rule
+ public MockitoRule rule = MockitoJUnit.rule().strictness(Strictness.STRICT_STUBS);
+
private static final String TOPIC = "topic";
- private static final String OTHER_TOPIC = "other-topic";
private static final Map<String, Object> PARTITION = Collections.singletonMap("key", "partition".getBytes());
private static final Map<String, Object> OFFSET = Collections.singletonMap("key", 12);
@@ -125,43 +142,55 @@ public class WorkerSourceTaskTest {
private static final byte[] SERIALIZED_KEY = "converted-key".getBytes();
private static final byte[] SERIALIZED_RECORD = "converted-record".getBytes();
- private ExecutorService executor = Executors.newSingleThreadExecutor();
- private ConnectorTaskId taskId = new ConnectorTaskId("job", 0);
- private ConnectorTaskId taskId1 = new ConnectorTaskId("job", 1);
+ private final ExecutorService executor = Executors.newSingleThreadExecutor();
+ private final ConnectorTaskId taskId = new ConnectorTaskId("job", 0);
private WorkerConfig config;
private SourceConnectorConfig sourceConfig;
private Plugins plugins;
private MockConnectMetrics metrics;
- @Mock private SourceTask sourceTask;
- @Mock private Converter keyConverter;
- @Mock private Converter valueConverter;
- @Mock private HeaderConverter headerConverter;
- @Mock private TransformationChain<SourceRecord> transformationChain;
- @Mock private KafkaProducer<byte[], byte[]> producer;
- @Mock private TopicAdmin admin;
- @Mock private CloseableOffsetStorageReader offsetReader;
- @Mock private OffsetStorageWriter offsetWriter;
- @Mock private ConnectorOffsetBackingStore offsetStore;
- @Mock private ClusterConfigState clusterConfigState;
+ @Mock
+ private SourceTask sourceTask;
+ @Mock
+ private Converter keyConverter;
+ @Mock
+ private Converter valueConverter;
+ @Mock
+ private HeaderConverter headerConverter;
+ @Mock
+ private TransformationChain<SourceRecord> transformationChain;
+ @Mock
+ private KafkaProducer<byte[], byte[]> producer;
+ @Mock
+ private TopicAdmin admin;
+ @Mock
+ private CloseableOffsetStorageReader offsetReader;
+ @Mock
+ private OffsetStorageWriter offsetWriter;
+ @Mock
+ private ConnectorOffsetBackingStore offsetStore;
+ @Mock
+ private ClusterConfigState clusterConfigState;
private WorkerSourceTask workerTask;
- @Mock private Future<RecordMetadata> sendFuture;
- @MockStrict private TaskStatus.Listener statusListener;
- @Mock private StatusBackingStore statusBackingStore;
- @Mock private ErrorHandlingMetrics errorHandlingMetrics;
-
- private Capture<org.apache.kafka.clients.producer.Callback> producerCallbacks;
+ @Mock
+ private TaskStatus.Listener statusListener;
+ @Mock
+ private StatusBackingStore statusBackingStore;
+ @Mock
+ private ErrorHandlingMetrics errorHandlingMetrics;
private static final Map<String, String> TASK_PROPS = new HashMap<>();
+
static {
TASK_PROPS.put(TaskConfig.TASK_CLASS_CONFIG, TestSourceTask.class.getName());
}
+
private static final TaskConfig TASK_CONFIG = new TaskConfig(TASK_PROPS);
- private static final List<SourceRecord> RECORDS = Arrays.asList(
+ private static final List<SourceRecord> RECORDS = Collections.singletonList(
new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD)
);
- private boolean enableTopicCreation;
+ private final boolean enableTopicCreation;
@ParameterizedTest.Parameters
public static Collection<Boolean> parameters() {
@@ -178,7 +207,6 @@ public class WorkerSourceTaskTest {
plugins = new Plugins(workerProps);
config = new StandaloneConfig(workerProps);
sourceConfig = new SourceConnectorConfig(plugins, sourceConnectorPropsWithGroups(TOPIC), true);
- producerCallbacks = EasyMock.newCapture();
metrics = new MockConnectMetrics();
}
@@ -212,6 +240,7 @@ public class WorkerSourceTaskTest {
@After
public void tearDown() {
if (metrics != null) metrics.stop();
+ verifyNoMoreInteractions(statusListener);
}
private void createWorkerTask() {
@@ -243,19 +272,10 @@ public class WorkerSourceTaskTest {
final CountDownLatch pauseLatch = new CountDownLatch(1);
createWorkerTask(TargetState.PAUSED);
-
- statusListener.onPause(taskId);
- EasyMock.expectLastCall().andAnswer(() -> {
+ doAnswer(invocation -> {
pauseLatch.countDown();
return null;
- });
-
- expectClose();
-
- statusListener.onShutdown(taskId);
- EasyMock.expectLastCall();
-
- PowerMock.replayAll();
+ }).when(statusListener).onPause(taskId);
workerTask.initialize(TASK_CONFIG);
Future<?> taskFuture = executor.submit(workerTask);
@@ -266,41 +286,25 @@ public class WorkerSourceTaskTest {
taskFuture.get();
- PowerMock.verifyAll();
+ verify(statusListener).onPause(taskId);
+ verify(statusListener).onShutdown(taskId);
+ verifyClose();
}
@Test
public void testPause() throws Exception {
createWorkerTask();
- expectCleanStartup();
-
AtomicInteger count = new AtomicInteger(0);
CountDownLatch pollLatch = expectPolls(10, count);
// In this test, we don't flush, so nothing goes any further than the offset writer
expectTopicCreation(TOPIC);
-
- statusListener.onPause(taskId);
- EasyMock.expectLastCall();
-
- sourceTask.stop();
- EasyMock.expectLastCall();
-
- offsetWriter.offset(PARTITION, OFFSET);
- PowerMock.expectLastCall();
- expectOffsetFlush(true);
-
- statusListener.onShutdown(taskId);
- EasyMock.expectLastCall();
-
- expectClose();
-
- PowerMock.replayAll();
+ expectOffsetFlush();
workerTask.initialize(TASK_CONFIG);
Future<?> taskFuture = executor.submit(workerTask);
- assertTrue(awaitLatch(pollLatch));
+ ConcurrencyUtils.awaitLatch(pollLatch, POLL_TIMEOUT_MSG);
workerTask.transitionTo(TargetState.PAUSED);
@@ -315,117 +319,90 @@ public class WorkerSourceTaskTest {
taskFuture.get();
- PowerMock.verifyAll();
+ verifyCleanStartup();
+ verifyTaskGetTopic(count.get());
+ verifyOffsetFlush(true);
+ verifyTopicCreation(TOPIC);
+ verify(statusListener).onPause(taskId);
+ verify(statusListener).onShutdown(taskId);
+ verify(sourceTask).stop();
+ verify(offsetWriter).offset(PARTITION, OFFSET);
+ verifyClose();
}
@Test
public void testPollsInBackground() throws Exception {
createWorkerTask();
- expectCleanStartup();
-
final CountDownLatch pollLatch = expectPolls(10);
- // In this test, we don't flush, so nothing goes any further than the offset writer
expectTopicCreation(TOPIC);
-
- sourceTask.stop();
- EasyMock.expectLastCall();
-
- offsetWriter.offset(PARTITION, OFFSET);
- PowerMock.expectLastCall();
- expectOffsetFlush(true);
-
- statusListener.onShutdown(taskId);
- EasyMock.expectLastCall();
-
- expectClose();
-
- PowerMock.replayAll();
+ expectOffsetFlush();
workerTask.initialize(TASK_CONFIG);
Future<?> taskFuture = executor.submit(workerTask);
- assertTrue(awaitLatch(pollLatch));
+ ConcurrencyUtils.awaitLatch(pollLatch, POLL_TIMEOUT_MSG);
workerTask.stop();
assertTrue(workerTask.awaitStop(1000));
taskFuture.get();
assertPollMetrics(10);
-
- PowerMock.verifyAll();
+ verifyCleanStartup();
+ verifyOffsetFlush(true);
+ verify(offsetWriter).offset(PARTITION, OFFSET);
+ verify(statusListener).onShutdown(taskId);
+ verifyClose();
}
@Test
public void testFailureInPoll() throws Exception {
createWorkerTask();
- expectCleanStartup();
-
final CountDownLatch pollLatch = new CountDownLatch(1);
final RuntimeException exception = new RuntimeException();
- EasyMock.expect(sourceTask.poll()).andAnswer(() -> {
+ when(sourceTask.poll()).thenAnswer(invocation -> {
pollLatch.countDown();
throw exception;
});
- statusListener.onFailure(taskId, exception);
- EasyMock.expectLastCall();
-
- sourceTask.stop();
- EasyMock.expectLastCall();
expectEmptyOffsetFlush();
- expectClose();
-
- PowerMock.replayAll();
-
workerTask.initialize(TASK_CONFIG);
Future<?> taskFuture = executor.submit(workerTask);
- assertTrue(awaitLatch(pollLatch));
+ ConcurrencyUtils.awaitLatch(pollLatch, POLL_TIMEOUT_MSG);
//Failure in poll should trigger automatic stop of the task
assertTrue(workerTask.awaitStop(1000));
- assertShouldSkipCommit();
taskFuture.get();
assertPollMetrics(0);
- PowerMock.verifyAll();
+ verifyCleanStartup();
+ verify(statusListener).onFailure(taskId, exception);
+ verify(sourceTask).stop();
+ assertShouldSkipCommit();
+ verifyOffsetFlush(true);
+ verifyClose();
}
@Test
public void testFailureInPollAfterCancel() throws Exception {
createWorkerTask();
- expectCleanStartup();
-
final CountDownLatch pollLatch = new CountDownLatch(1);
final CountDownLatch workerCancelLatch = new CountDownLatch(1);
final RuntimeException exception = new RuntimeException();
- EasyMock.expect(sourceTask.poll()).andAnswer(() -> {
+ when(sourceTask.poll()).thenAnswer(invocation -> {
pollLatch.countDown();
- assertTrue(awaitLatch(workerCancelLatch));
+ ConcurrencyUtils.awaitLatch(workerCancelLatch, "Timeout waiting for main test thread to cancel task.");
throw exception;
});
- offsetReader.close();
- PowerMock.expectLastCall();
-
- producer.close(Duration.ZERO);
- PowerMock.expectLastCall();
-
- sourceTask.stop();
- EasyMock.expectLastCall();
-
- expectClose();
-
- PowerMock.replayAll();
-
workerTask.initialize(TASK_CONFIG);
Future<?> taskFuture = executor.submit(workerTask);
- assertTrue(awaitLatch(pollLatch));
+ ConcurrencyUtils.awaitLatch(pollLatch, POLL_TIMEOUT_MSG);
workerTask.cancel();
workerCancelLatch.countDown();
assertTrue(workerTask.awaitStop(1000));
@@ -433,39 +410,39 @@ public class WorkerSourceTaskTest {
taskFuture.get();
assertPollMetrics(0);
- PowerMock.verifyAll();
+ verifyCleanStartup();
+ verify(offsetReader, atLeastOnce()).close();
+ verify(producer).close(Duration.ZERO);
+ verify(sourceTask).stop();
+ verify(admin).close(any(Duration.class));
+ verify(transformationChain).close();
+ verify(offsetStore).stop();
+
+ try {
+ verify(headerConverter).close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
}
@Test
public void testFailureInPollAfterStop() throws Exception {
createWorkerTask();
- expectCleanStartup();
-
final CountDownLatch pollLatch = new CountDownLatch(1);
final CountDownLatch workerStopLatch = new CountDownLatch(1);
final RuntimeException exception = new RuntimeException();
- EasyMock.expect(sourceTask.poll()).andAnswer(() -> {
+ when(sourceTask.poll()).thenAnswer(invocation -> {
pollLatch.countDown();
- assertTrue(awaitLatch(workerStopLatch));
+ ConcurrencyUtils.awaitLatch(workerStopLatch, "Timeout waiting for main test thread to stop task");
throw exception;
});
-
- statusListener.onShutdown(taskId);
- EasyMock.expectLastCall();
-
- sourceTask.stop();
- EasyMock.expectLastCall();
- expectOffsetFlush(true);
-
- expectClose();
-
- PowerMock.replayAll();
+ expectOffsetFlush();
workerTask.initialize(TASK_CONFIG);
Future<?> taskFuture = executor.submit(workerTask);
- assertTrue(awaitLatch(pollLatch));
+ ConcurrencyUtils.awaitLatch(pollLatch, POLL_TIMEOUT_MSG);
workerTask.stop();
workerStopLatch.countDown();
assertTrue(workerTask.awaitStop(1000));
@@ -474,7 +451,11 @@ public class WorkerSourceTaskTest {
taskFuture.get();
assertPollMetrics(0);
- PowerMock.verifyAll();
+ verifyCleanStartup();
+ verify(statusListener).onShutdown(taskId);
+ verify(sourceTask).stop();
+ verifyOffsetFlush(true);
+ verifyClose();
}
@Test
@@ -482,35 +463,29 @@ public class WorkerSourceTaskTest {
// Test that the task handles an empty list of records
createWorkerTask();
- expectCleanStartup();
-
// We'll wait for some data, then trigger a flush
- final CountDownLatch pollLatch = expectEmptyPolls(1, new AtomicInteger());
+ final CountDownLatch pollLatch = expectEmptyPolls(new AtomicInteger());
expectEmptyOffsetFlush();
- sourceTask.stop();
- EasyMock.expectLastCall();
- expectEmptyOffsetFlush();
-
- statusListener.onShutdown(taskId);
- EasyMock.expectLastCall();
-
- expectClose();
-
- PowerMock.replayAll();
-
workerTask.initialize(TASK_CONFIG);
Future<?> taskFuture = executor.submit(workerTask);
- assertTrue(awaitLatch(pollLatch));
+ ConcurrencyUtils.awaitLatch(pollLatch, POLL_TIMEOUT_MSG);
assertTrue(workerTask.commitOffsets());
+ verify(offsetWriter).beginFlush(anyLong(), any(TimeUnit.class));
+
workerTask.stop();
assertTrue(workerTask.awaitStop(1000));
+ verify(offsetWriter, times(2)).beginFlush(anyLong(), any(TimeUnit.class));
+ verifyNoMoreInteractions(offsetWriter);
taskFuture.get();
assertPollMetrics(0);
- PowerMock.verifyAll();
+ verifyCleanStartup();
+ verify(sourceTask).stop();
+ verify(statusListener).onShutdown(taskId);
+ verifyClose();
}
@Test
@@ -518,40 +493,33 @@ public class WorkerSourceTaskTest {
// Test that the task commits properly when prompted
createWorkerTask();
- expectCleanStartup();
-
// We'll wait for some data, then trigger a flush
final CountDownLatch pollLatch = expectPolls(1);
- expectOffsetFlush(true);
-
- offsetWriter.offset(PARTITION, OFFSET);
- PowerMock.expectLastCall().atLeastOnce();
expectTopicCreation(TOPIC);
-
- sourceTask.stop();
- EasyMock.expectLastCall();
- expectEmptyOffsetFlush();
-
- statusListener.onShutdown(taskId);
- EasyMock.expectLastCall();
-
- expectClose();
-
- PowerMock.replayAll();
+ expectBeginFlush(Arrays.asList(true, false).iterator()::next);
+ expectOffsetFlush(true, true);
workerTask.initialize(TASK_CONFIG);
Future<?> taskFuture = executor.submit(workerTask);
- assertTrue(awaitLatch(pollLatch));
+ ConcurrencyUtils.awaitLatch(pollLatch, POLL_TIMEOUT_MSG);
assertTrue(workerTask.commitOffsets());
+
workerTask.stop();
assertTrue(workerTask.awaitStop(1000));
taskFuture.get();
assertPollMetrics(1);
- PowerMock.verifyAll();
+ verifyCleanStartup();
+ verifyTopicCreation(TOPIC);
+ verify(offsetWriter, times(2)).beginFlush(anyLong(), any(TimeUnit.class));
+ verify(offsetWriter, atLeastOnce()).offset(PARTITION, OFFSET);
+ verify(sourceTask).stop();
+ verify(statusListener).onShutdown(taskId);
+ verifyOffsetFlush(true, 2);
+ verifyClose();
}
@Test
@@ -559,78 +527,70 @@ public class WorkerSourceTaskTest {
// Test that the task commits properly when prompted
createWorkerTask();
- expectCleanStartup();
-
// We'll wait for some data, then trigger a flush
final CountDownLatch pollLatch = expectPolls(1);
- expectOffsetFlush(true);
-
- offsetWriter.offset(PARTITION, OFFSET);
- PowerMock.expectLastCall().atLeastOnce();
+ expectBeginFlush();
+ expectOffsetFlush(true, false);
expectTopicCreation(TOPIC);
- sourceTask.stop();
- EasyMock.expectLastCall();
- expectOffsetFlush(false);
-
- statusListener.onShutdown(taskId);
- EasyMock.expectLastCall();
-
- expectClose();
-
- PowerMock.replayAll();
-
workerTask.initialize(TASK_CONFIG);
Future<?> taskFuture = executor.submit(workerTask);
- assertTrue(awaitLatch(pollLatch));
+ ConcurrencyUtils.awaitLatch(pollLatch, POLL_TIMEOUT_MSG);
assertTrue(workerTask.commitOffsets());
+
workerTask.stop();
assertTrue(workerTask.awaitStop(1000));
taskFuture.get();
assertPollMetrics(1);
- PowerMock.verifyAll();
+ verifyCleanStartup();
+ verify(sourceTask).stop();
+ verify(offsetWriter, atLeastOnce()).offset(PARTITION, OFFSET);
+ verify(statusListener).onShutdown(taskId);
+
+ verifyOffsetFlush(true); // First call to doFlush() succeeded
+ verifyOffsetFlush(false); // Second call threw a TimeoutException
+ verifyClose();
}
@Test
- public void testSendRecordsRetries() throws Exception {
+ public void testSendRecordsRetries() {
createWorkerTask();
- // Differentiate only by Kafka partition so we can reuse conversion expectations
+ // Differentiate only by Kafka partition, so we can reuse conversion expectations
SourceRecord record1 = new SourceRecord(PARTITION, OFFSET, "topic", 1, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);
SourceRecord record2 = new SourceRecord(PARTITION, OFFSET, "topic", 2, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);
SourceRecord record3 = new SourceRecord(PARTITION, OFFSET, "topic", 3, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);
expectTopicCreation(TOPIC);
+ expectPreliminaryCalls();
- // First round
- expectSendRecordOnce();
- // Any Producer retriable exception should work here
- expectSendRecordSyncFailure(new org.apache.kafka.common.errors.TimeoutException("retriable sync failure"));
-
- // Second round
- expectSendRecordOnce();
- expectSendRecordOnce();
+ expectTaskGetTopic();
- PowerMock.replayAll();
+ when(producer.send(any(ProducerRecord.class), any(Callback.class)))
+ // First round
+ .thenAnswer(producerSendAnswer(true))
+ // Any Producer retriable exception should work here
+ .thenThrow(new org.apache.kafka.common.errors.TimeoutException("retriable sync failure"))
+ // Second round
+ .thenAnswer(producerSendAnswer(true))
+ .thenAnswer(producerSendAnswer(true));
// Try to send 3, make first pass, second fail. Should save last two
- Whitebox.setInternalState(workerTask, "toSend", Arrays.asList(record1, record2, record3));
- Whitebox.invokeMethod(workerTask, "sendRecords");
- assertEquals(Arrays.asList(record2, record3), Whitebox.getInternalState(workerTask, "toSend"));
+ workerTask.toSend = Arrays.asList(record1, record2, record3);
+ workerTask.sendRecords();
+ assertEquals(Arrays.asList(record2, record3), workerTask.toSend);
// Next they all succeed
- Whitebox.invokeMethod(workerTask, "sendRecords");
- assertNull(Whitebox.getInternalState(workerTask, "toSend"));
-
- PowerMock.verifyAll();
+ workerTask.sendRecords();
+ assertNull(workerTask.toSend);
}
@Test
- public void testSendRecordsProducerCallbackFail() throws Exception {
+ public void testSendRecordsProducerCallbackFail() {
createWorkerTask();
SourceRecord record1 = new SourceRecord(PARTITION, OFFSET, "topic", 1, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);
@@ -639,13 +599,13 @@ public class WorkerSourceTaskTest {
expectTopicCreation(TOPIC);
expectSendRecordProducerCallbackFail();
- expectApplyTransformationChain(false);
- expectConvertHeadersAndKeyValue(false);
- PowerMock.replayAll();
+ workerTask.toSend = Arrays.asList(record1, record2);
+ assertThrows(ConnectException.class, () -> workerTask.sendRecords());
- Whitebox.setInternalState(workerTask, "toSend", Arrays.asList(record1, record2));
- assertThrows(ConnectException.class, () -> Whitebox.invokeMethod(workerTask, "sendRecords"));
+ verify(transformationChain, times(2)).apply(any(SourceRecord.class));
+ verify(keyConverter, times(2)).fromConnectData(anyString(), any(Headers.class), eq(KEY_SCHEMA), eq(KEY));
+ verify(valueConverter, times(2)).fromConnectData(anyString(), any(Headers.class), eq(RECORD_SCHEMA), eq(RECORD));
}
@Test
@@ -658,38 +618,34 @@ public class WorkerSourceTaskTest {
expectPreliminaryCalls();
expectTopicCreation(TOPIC);
- EasyMock.expect(producer.send(EasyMock.anyObject(), EasyMock.anyObject()))
- .andThrow(new KafkaException("Producer closed while send in progress", new InvalidTopicException(TOPIC)));
+ when(producer.send(any(ProducerRecord.class), any(Callback.class)))
+ .thenThrow(new KafkaException("Producer closed while send in progress", new InvalidTopicException(TOPIC)));
- PowerMock.replayAll();
-
- Whitebox.setInternalState(workerTask, "toSend", Arrays.asList(record1, record2));
- assertThrows(ConnectException.class, () -> Whitebox.invokeMethod(workerTask, "sendRecords"));
+ workerTask.toSend = Arrays.asList(record1, record2);
+ assertThrows(ConnectException.class, () -> workerTask.sendRecords());
}
@Test
public void testSendRecordsTaskCommitRecordFail() throws Exception {
createWorkerTask();
- // Differentiate only by Kafka partition so we can reuse conversion expectations
+ // Differentiate only by Kafka partition, so we can reuse conversion expectations
SourceRecord record1 = new SourceRecord(PARTITION, OFFSET, "topic", 1, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);
SourceRecord record2 = new SourceRecord(PARTITION, OFFSET, "topic", 2, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);
SourceRecord record3 = new SourceRecord(PARTITION, OFFSET, "topic", 3, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);
expectTopicCreation(TOPIC);
+ expectSendRecord();
// Source task commit record failure will not cause the task to abort
- expectSendRecordOnce();
- expectSendRecordTaskCommitRecordFail(false);
- expectSendRecordOnce();
-
- PowerMock.replayAll();
+ doNothing()
+ .doThrow(new RuntimeException("Error committing record in source task"))
+ .doNothing()
+ .when(sourceTask).commitRecord(any(SourceRecord.class), any(RecordMetadata.class));
- Whitebox.setInternalState(workerTask, "toSend", Arrays.asList(record1, record2, record3));
- Whitebox.invokeMethod(workerTask, "sendRecords");
- assertNull(Whitebox.getInternalState(workerTask, "toSend"));
-
- PowerMock.verifyAll();
+ workerTask.toSend = Arrays.asList(record1, record2, record3);
+ workerTask.sendRecords();
+ assertNull(workerTask.toSend);
}
@Test
@@ -697,7 +653,7 @@ public class WorkerSourceTaskTest {
createWorkerTaskWithErrorToleration();
expectTopicCreation(TOPIC);
- //Use different offsets for each record so we can verify all were committed
+ //Use different offsets for each record, so we can verify all were committed
final Map<String, Object> offset2 = Collections.singletonMap("key", 13);
// send two records
@@ -706,30 +662,28 @@ public class WorkerSourceTaskTest {
// and no ConnectException will be thrown
SourceRecord record1 = new SourceRecord(PARTITION, OFFSET, TOPIC, 1, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);
SourceRecord record2 = new SourceRecord(PARTITION, offset2, TOPIC, 2, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);
- expectOffsetFlush(true);
- expectSendRecordOnce();
- expectSendRecordProducerCallbackFail();
- sourceTask.commitRecord(EasyMock.anyObject(SourceRecord.class), EasyMock.isNull());
- //As of KAFKA-14079 all offsets should be committed, even for failed records (if ignored)
- //Only the last offset will be passed to the method as everything up to that point is committed
- //Before KAFKA-14079 offset 12 would have been passed and not 13 as it would have been unacked
- offsetWriter.offset(PARTITION, offset2);
- PowerMock.expectLastCall();
+ expectOffsetFlush();
+ expectPreliminaryCalls();
- PowerMock.replayAll();
+ when(producer.send(any(ProducerRecord.class), any(Callback.class)))
+ .thenAnswer(producerSendAnswer(true))
+ .thenAnswer(producerSendAnswer(false));
//Send records and then commit offsets and verify both were committed and no exception
- Whitebox.setInternalState(workerTask, "toSend", Arrays.asList(record1, record2));
- Whitebox.invokeMethod(workerTask, "sendRecords");
- Whitebox.invokeMethod(workerTask, "updateCommittableOffsets");
+ workerTask.toSend = Arrays.asList(record1, record2);
+ workerTask.sendRecords();
+ workerTask.updateCommittableOffsets();
workerTask.commitOffsets();
- PowerMock.verifyAll();
+ //As of KAFKA-14079 all offsets should be committed, even for failed records (if ignored)
+ //Only the last offset will be passed to the method as everything up to that point is committed
+ //Before KAFKA-14079 offset 12 would have been passed and not 13 as it would have been unacked
+ verify(offsetWriter).offset(PARTITION, offset2);
+ verify(sourceTask).commitRecord(any(SourceRecord.class), isNull());
//Double check to make sure all submitted records were cleared
- assertEquals(0, ((SubmittedRecords) Whitebox.getInternalState(workerTask,
- "submittedRecords")).records.size());
+ assertEquals(0, workerTask.submittedRecords.records.size());
}
@Test
@@ -739,30 +693,13 @@ public class WorkerSourceTaskTest {
createWorkerTask();
- offsetStore.start();
- EasyMock.expectLastCall();
- sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
- EasyMock.expectLastCall();
- sourceTask.start(TASK_PROPS);
- EasyMock.expectLastCall().andAnswer(() -> {
+ doAnswer((Answer<Object>) invocation -> {
startupLatch.countDown();
- assertTrue(awaitLatch(finishStartupLatch));
+ ConcurrencyUtils.awaitLatch(finishStartupLatch, "Timeout waiting for main test thread to allow task startup to complete");
return null;
- });
-
- statusListener.onStartup(taskId);
- EasyMock.expectLastCall();
-
- sourceTask.stop();
- EasyMock.expectLastCall();
- expectOffsetFlush(true);
-
- statusListener.onShutdown(taskId);
- EasyMock.expectLastCall();
+ }).when(sourceTask).start(TASK_PROPS);
- expectClose();
-
- PowerMock.replayAll();
+ expectOffsetFlush();
workerTask.initialize(TASK_CONFIG);
Future<?> workerTaskFuture = executor.submit(workerTask);
@@ -770,31 +707,28 @@ public class WorkerSourceTaskTest {
// Stopping immediately while the other thread has work to do should result in no polling, no offset commits,
// exiting the work thread immediately, and the stop() method will be invoked in the background thread since it
// cannot be invoked immediately in the thread trying to stop the task.
- assertTrue(awaitLatch(startupLatch));
+ ConcurrencyUtils.awaitLatch(startupLatch, "Timeout waiting for task to begin startup");
workerTask.stop();
finishStartupLatch.countDown();
assertTrue(workerTask.awaitStop(1000));
workerTaskFuture.get();
-
- PowerMock.verifyAll();
+ verify(offsetStore).start();
+ verify(sourceTask).initialize(any(SourceTaskContext.class));
+ verify(sourceTask).start(TASK_PROPS);
+ verify(statusListener).onStartup(taskId);
+ verify(statusListener).onShutdown(taskId);
+ verify(sourceTask).stop();
+ verifyClose();
}
@Test
public void testCancel() {
createWorkerTask();
- offsetReader.close();
- PowerMock.expectLastCall();
-
- producer.close(Duration.ZERO);
- PowerMock.expectLastCall();
-
- PowerMock.replayAll();
-
workerTask.cancel();
-
- PowerMock.verifyAll();
+ verify(offsetReader).close();
+ verify(producer).close(Duration.ZERO);
}
private TopicAdmin.TopicCreationResponse createdTopic(String topic) {
@@ -804,26 +738,21 @@ public class WorkerSourceTaskTest {
}
private void expectPreliminaryCalls() {
- expectPreliminaryCalls(TOPIC);
- }
-
- private void expectPreliminaryCalls(String topic) {
- expectConvertHeadersAndKeyValue(topic, true, emptyHeaders());
- expectApplyTransformationChain(false);
+ expectConvertHeadersAndKeyValue(TOPIC, emptyHeaders());
+ expectApplyTransformationChain();
}
- private CountDownLatch expectEmptyPolls(int minimum, final AtomicInteger count) throws InterruptedException {
- final CountDownLatch latch = new CountDownLatch(minimum);
+ private CountDownLatch expectEmptyPolls(final AtomicInteger count) throws InterruptedException {
+ final CountDownLatch latch = new CountDownLatch(1);
// 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(() -> {
- count.incrementAndGet();
- latch.countDown();
- Thread.sleep(10);
- return Collections.emptyList();
- });
+ when(sourceTask.poll()).thenAnswer((Answer<List<SourceRecord>>) invocation -> {
+ count.incrementAndGet();
+ latch.countDown();
+ Thread.sleep(10);
+ return Collections.emptyList();
+ });
return latch;
}
@@ -832,15 +761,15 @@ public class WorkerSourceTaskTest {
// 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(() -> {
- count.incrementAndGet();
- latch.countDown();
- Thread.sleep(10);
- return RECORDS;
- });
+ doAnswer((Answer<List<SourceRecord>>) invocation -> {
+ count.incrementAndGet();
+ latch.countDown();
+ Thread.sleep(10);
+ return RECORDS;
+ }).when(sourceTask).poll();
+
// Fallout of the poll() call
- expectSendRecordAnyTimes();
+ expectSendRecord();
return latch;
}
@@ -848,186 +777,131 @@ public class WorkerSourceTaskTest {
return expectPolls(count, new AtomicInteger());
}
- @SuppressWarnings("unchecked")
- private void expectSendRecordSyncFailure(Throwable error) {
- expectConvertHeadersAndKeyValue(false);
- expectApplyTransformationChain(false);
-
- EasyMock.expect(
- producer.send(EasyMock.anyObject(ProducerRecord.class),
- EasyMock.anyObject(org.apache.kafka.clients.producer.Callback.class)))
- .andThrow(error);
+ private void expectSendRecord() {
+ expectSendRecordTaskCommitRecordSucceed();
}
- private Capture<ProducerRecord<byte[], byte[]>> expectSendRecordAnyTimes() throws InterruptedException {
- return expectSendRecordTaskCommitRecordSucceed(true);
+ private void expectSendRecordProducerCallbackFail() {
+ expectSendRecord(TOPIC, false, emptyHeaders());
}
- private Capture<ProducerRecord<byte[], byte[]>> expectSendRecordOnce() throws InterruptedException {
- return expectSendRecordTaskCommitRecordSucceed(false);
+ private void expectSendRecordTaskCommitRecordSucceed() {
+ expectSendRecord(TOPIC, true, emptyHeaders());
}
- private Capture<ProducerRecord<byte[], byte[]>> expectSendRecordProducerCallbackFail() throws InterruptedException {
- return expectSendRecord(TOPIC, false, false, false, true, emptyHeaders());
- }
+ private void expectSendRecord(String topic, boolean sendSuccess, Headers headers) {
+ expectConvertHeadersAndKeyValue(topic, headers);
- private Capture<ProducerRecord<byte[], byte[]>> expectSendRecordTaskCommitRecordSucceed(boolean anyTimes) throws InterruptedException {
- return expectSendRecord(TOPIC, anyTimes, true, true, true, emptyHeaders());
- }
+ expectApplyTransformationChain();
- private Capture<ProducerRecord<byte[], byte[]>> expectSendRecordTaskCommitRecordFail(boolean anyTimes) throws InterruptedException {
- return expectSendRecord(TOPIC, anyTimes, true, false, true, emptyHeaders());
- }
-
- private Capture<ProducerRecord<byte[], byte[]>> expectSendRecord(
- String topic,
- boolean anyTimes,
- boolean sendSuccess,
- boolean commitSuccess,
- boolean isMockedConverters,
- Headers headers
- ) throws InterruptedException {
- if (isMockedConverters) {
- expectConvertHeadersAndKeyValue(topic, anyTimes, headers);
+ if (sendSuccess) {
+ // 2. As a result of a successful producer send callback, we'll notify the source task of the record commit
+ expectTaskGetTopic();
}
- expectApplyTransformationChain(anyTimes);
-
- Capture<ProducerRecord<byte[], byte[]>> sent = EasyMock.newCapture();
-
- // 1. Converted data passed to the producer, which will need callbacks invoked for flush to work
- IExpectationSetters<Future<RecordMetadata>> expect = EasyMock.expect(
- producer.send(EasyMock.capture(sent),
- EasyMock.capture(producerCallbacks)));
- IAnswer<Future<RecordMetadata>> expectResponse = () -> {
- synchronized (producerCallbacks) {
- for (org.apache.kafka.clients.producer.Callback cb : producerCallbacks.getValues()) {
- if (sendSuccess) {
- cb.onCompletion(new RecordMetadata(new TopicPartition("foo", 0), 0, 0,
- 0L, 0, 0), null);
- } else {
- cb.onCompletion(null, new TopicAuthorizationException("foo"));
- }
- }
- producerCallbacks.reset();
+ doAnswer(producerSendAnswer(sendSuccess))
+ .when(producer).send(any(ProducerRecord.class), any(Callback.class));
+ }
+
+ private Answer<Future<RecordMetadata>> producerSendAnswer(boolean sendSuccess) {
+ return invocation -> {
+ Callback cb = invocation.getArgument(1);
+ if (sendSuccess) {
+ cb.onCompletion(new RecordMetadata(new TopicPartition("foo", 0), 0, 0, 0L, 0, 0),
+ null);
+ } else {
+ cb.onCompletion(null, new TopicAuthorizationException("foo"));
}
- return sendFuture;
+
+ return null;
};
- if (anyTimes)
- expect.andStubAnswer(expectResponse);
- else
- expect.andAnswer(expectResponse);
+ }
- if (sendSuccess) {
- // 2. As a result of a successful producer send callback, we'll notify the source task of the record commit
- expectTaskCommitRecordWithOffset(anyTimes, commitSuccess);
- expectTaskGetTopic(anyTimes);
+ private void expectConvertHeadersAndKeyValue(String topic, Headers headers) {
+ if (headers.iterator().hasNext()) {
+ when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA),
+ anyString()))
+ .thenAnswer((Answer<byte[]>) invocation -> {
+ String headerValue = invocation.getArgument(3, String.class);
+ return headerValue.getBytes(StandardCharsets.UTF_8);
+ });
}
- return sent;
+ when(keyConverter.fromConnectData(eq(topic), any(Headers.class), eq(KEY_SCHEMA), eq(KEY)))
+ .thenReturn(SERIALIZED_KEY);
+ when(valueConverter.fromConnectData(eq(topic), any(Headers.class), eq(RECORD_SCHEMA),
+ eq(RECORD)))
+ .thenReturn(SERIALIZED_RECORD);
}
- private void expectConvertHeadersAndKeyValue(boolean anyTimes) {
- expectConvertHeadersAndKeyValue(TOPIC, anyTimes, emptyHeaders());
+ private void expectApplyTransformationChain() {
+ when(transformationChain.apply(any(SourceRecord.class)))
+ .thenAnswer(AdditionalAnswers.returnsFirstArg());
}
- private void expectConvertHeadersAndKeyValue(String topic, boolean anyTimes, Headers headers) {
- for (Header header : headers) {
- IExpectationSetters<byte[]> convertHeaderExpect = EasyMock.expect(headerConverter.fromConnectHeader(topic, header.key(), Schema.STRING_SCHEMA, new String(header.value())));
- if (anyTimes)
- convertHeaderExpect.andStubReturn(header.value());
- else
- convertHeaderExpect.andReturn(header.value());
- }
- IExpectationSetters<byte[]> convertKeyExpect = EasyMock.expect(keyConverter.fromConnectData(topic, headers, KEY_SCHEMA, KEY));
- if (anyTimes)
- convertKeyExpect.andStubReturn(SERIALIZED_KEY);
- else
- convertKeyExpect.andReturn(SERIALIZED_KEY);
- IExpectationSetters<byte[]> convertValueExpect = EasyMock.expect(valueConverter.fromConnectData(topic, headers, RECORD_SCHEMA, RECORD));
- if (anyTimes)
- convertValueExpect.andStubReturn(SERIALIZED_RECORD);
- else
- convertValueExpect.andReturn(SERIALIZED_RECORD);
+ private void expectTaskGetTopic() {
+ when(statusBackingStore.getTopic(anyString(), anyString())).thenAnswer((Answer<TopicStatus>) invocation -> {
+ String connector = invocation.getArgument(0, String.class);
+ String topic = invocation.getArgument(1, String.class);
+ return new TopicStatus(topic, new ConnectorTaskId(connector, 0), Time.SYSTEM.milliseconds());
+ });
}
- private void expectApplyTransformationChain(boolean anyTimes) {
- final Capture<SourceRecord> recordCapture = EasyMock.newCapture();
- IExpectationSetters<SourceRecord> convertKeyExpect = EasyMock.expect(transformationChain.apply(EasyMock.capture(recordCapture)));
- if (anyTimes)
- convertKeyExpect.andStubAnswer(recordCapture::getValue);
- else
- convertKeyExpect.andAnswer(recordCapture::getValue);
+ private void verifyTaskGetTopic(int times) {
+ ArgumentCaptor<String> connectorCapture = ArgumentCaptor.forClass(String.class);
+ ArgumentCaptor<String> topicCapture = ArgumentCaptor.forClass(String.class);
+ verify(statusBackingStore, times(times)).getTopic(connectorCapture.capture(), topicCapture.capture());
+
+ assertEquals("job", connectorCapture.getValue());
+ assertEquals(TOPIC, topicCapture.getValue());
}
- private void expectTaskCommitRecordWithOffset(boolean anyTimes, boolean succeed) throws InterruptedException {
- sourceTask.commitRecord(EasyMock.anyObject(SourceRecord.class), EasyMock.anyObject(RecordMetadata.class));
- IExpectationSetters<Void> expect = EasyMock.expectLastCall();
- if (!succeed) {
- expect = expect.andThrow(new RuntimeException("Error committing record in source task"));
- }
- if (anyTimes) {
- expect.anyTimes();
- }
+ private void expectBeginFlush() throws Exception {
+ expectBeginFlush(() -> true);
}
- private void expectTaskGetTopic(boolean anyTimes) {
- final Capture<String> connectorCapture = EasyMock.newCapture();
- final Capture<String> topicCapture = EasyMock.newCapture();
- IExpectationSetters<TopicStatus> expect = EasyMock.expect(statusBackingStore.getTopic(
- EasyMock.capture(connectorCapture),
- EasyMock.capture(topicCapture)));
- if (anyTimes) {
- expect.andStubAnswer(() -> new TopicStatus(
- topicCapture.getValue(),
- new ConnectorTaskId(connectorCapture.getValue(), 0),
- Time.SYSTEM.milliseconds()));
- } else {
- expect.andAnswer(() -> new TopicStatus(
- topicCapture.getValue(),
- new ConnectorTaskId(connectorCapture.getValue(), 0),
- Time.SYSTEM.milliseconds()));
- }
- if (connectorCapture.hasCaptured() && topicCapture.hasCaptured()) {
- assertEquals("job", connectorCapture.getValue());
- assertEquals(TOPIC, topicCapture.getValue());
- }
+ private void expectBeginFlush(Supplier<Boolean> resultSupplier) throws Exception {
+ when(offsetWriter.beginFlush(anyLong(), any(TimeUnit.class))).thenAnswer(ignored -> resultSupplier.get());
}
- private boolean awaitLatch(CountDownLatch latch) {
- try {
- return latch.await(5000, TimeUnit.MILLISECONDS);
- } catch (InterruptedException e) {
- // ignore
- }
- return false;
+ private void expectOffsetFlush() throws Exception {
+ expectBeginFlush();
+ expectOffsetFlush(true);
}
@SuppressWarnings("unchecked")
- private void expectOffsetFlush(boolean succeed) throws Exception {
- EasyMock.expect(offsetWriter.beginFlush(EasyMock.anyLong(), EasyMock.anyObject())).andReturn(true);
- Future<Void> flushFuture = PowerMock.createMock(Future.class);
- EasyMock.expect(offsetWriter.doFlush(EasyMock.anyObject(Callback.class))).andReturn(flushFuture);
+ private void expectOffsetFlush(Boolean... succeedList) throws Exception {
+ Future<Void> flushFuture = mock(Future.class);
+ when(offsetWriter.doFlush(any(org.apache.kafka.connect.util.Callback.class))).thenReturn(flushFuture);
+ LinkedList<Boolean> succeedQueue = new LinkedList<>(Arrays.asList(succeedList));
+
+ doAnswer(invocationOnMock -> {
+ boolean succeed = succeedQueue.pop();
+ if (succeed) {
+ return null;
+ } else {
+ throw new TimeoutException();
+ }
+ }).when(flushFuture).get(anyLong(), any(TimeUnit.class));
+ }
+
+ private void expectEmptyOffsetFlush() throws Exception {
+ expectBeginFlush(() -> false);
+ }
+
+ private void verifyOffsetFlush(boolean succeed) throws Exception {
+ verifyOffsetFlush(succeed, 1);
+ }
+
+ private void verifyOffsetFlush(boolean succeed, int times) throws Exception {
// Should throw for failure
- IExpectationSetters<Void> futureGetExpect = EasyMock.expect(
- flushFuture.get(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class)));
if (succeed) {
- sourceTask.commit();
- EasyMock.expectLastCall();
- futureGetExpect.andReturn(null);
+ verify(sourceTask, atLeast(times)).commit();
} else {
- futureGetExpect.andThrow(new TimeoutException());
- offsetWriter.cancelFlush();
- PowerMock.expectLastCall();
+ verify(offsetWriter, atLeast(times)).cancelFlush();
}
}
- private void expectEmptyOffsetFlush() throws Exception {
- EasyMock.expect(offsetWriter.beginFlush(EasyMock.anyLong(), EasyMock.anyObject())).andReturn(false);
- sourceTask.commit();
- EasyMock.expectLastCall();
- }
-
private void assertPollMetrics(int minimumPollCountExpected) {
MetricGroup sourceTaskGroup = workerTask.sourceTaskMetricsGroup().metricGroup();
MetricGroup taskGroup = workerTask.taskMetricsGroup().metricGroup();
@@ -1056,6 +930,7 @@ public class WorkerSourceTaskTest {
if (minimumPollCountExpected > 0) {
assertTrue(pollBatchTimeMax >= 0.0d);
}
+
assertTrue(Double.isNaN(pollBatchTimeAvg) || pollBatchTimeAvg > 0.0d);
double activeCount = metrics.currentMetricValueAsDouble(sourceTaskGroup, "source-record-active-count");
double activeCountMax = metrics.currentMetricValueAsDouble(sourceTaskGroup, "source-record-active-count-max");
@@ -1072,46 +947,43 @@ public class WorkerSourceTaskTest {
private abstract static class TestSourceTask extends SourceTask {
}
- private void expectCleanStartup() {
- offsetStore.start();
- EasyMock.expectLastCall();
- sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
- EasyMock.expectLastCall();
- sourceTask.start(TASK_PROPS);
- EasyMock.expectLastCall();
- statusListener.onStartup(taskId);
- EasyMock.expectLastCall();
+ private void verifyCleanStartup() {
+ verify(offsetStore).start();
+ verify(sourceTask).initialize(any(SourceTaskContext.class));
+ verify(sourceTask).start(TASK_PROPS);
+ verify(statusListener).onStartup(taskId);
}
- private void expectClose() {
- producer.close(EasyMock.anyObject(Duration.class));
- EasyMock.expectLastCall();
-
- admin.close(EasyMock.anyObject(Duration.class));
- EasyMock.expectLastCall();
-
- transformationChain.close();
- EasyMock.expectLastCall();
-
- offsetReader.close();
- EasyMock.expectLastCall();
-
- offsetStore.stop();
- EasyMock.expectLastCall();
+ private void verifyClose() {
+ verify(producer).close(any(Duration.class));
+ verify(admin).close(any(Duration.class));
+ verify(transformationChain).close();
+ verify(offsetReader).close();
+ verify(offsetStore).stop();
try {
- headerConverter.close();
+ verify(headerConverter).close();
} catch (IOException e) {
throw new RuntimeException(e);
}
- EasyMock.expectLastCall();
}
private void expectTopicCreation(String topic) {
if (config.topicCreationEnable()) {
- EasyMock.expect(admin.describeTopics(topic)).andReturn(Collections.emptyMap());
- Capture<NewTopic> newTopicCapture = EasyMock.newCapture();
- EasyMock.expect(admin.createOrFindTopics(EasyMock.capture(newTopicCapture))).andReturn(createdTopic(topic));
+ when(admin.describeTopics(topic)).thenReturn(Collections.emptyMap());
+ when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(topic));
+ }
+ }
+
+ private void verifyTopicCreation(String... topics) {
+ if (config.topicCreationEnable()) {
+ ArgumentCaptor<NewTopic> newTopicCapture = ArgumentCaptor.forClass(NewTopic.class);
+
+ verify(admin).createOrFindTopics(newTopicCapture.capture());
+ assertArrayEquals(topics, newTopicCapture.getAllValues()
+ .stream()
+ .map(NewTopic::name)
+ .toArray(String[]::new));
}
}