From 534f8c7178ca93f74135b5bc45c6cc90e9a2daa2 Mon Sep 17 00:00:00 2001 From: Hector Geraldino Date: Mon, 30 Jan 2023 23:16:28 -0500 Subject: [PATCH 1/7] Replace PowerMock with Mockito --- build.gradle | 2 +- .../runtime/AbstractWorkerSourceTaskTest.java | 724 ++++++++++-------- 2 files changed, 394 insertions(+), 332 deletions(-) diff --git a/build.gradle b/build.gradle index 8441ddeaf051f..68f91ed38ce51 100644 --- a/build.gradle +++ b/build.gradle @@ -414,7 +414,7 @@ subprojects { "**/KafkaConfigBackingStoreTest.*", "**/KafkaBasedLogTest.*", "**/StandaloneHerderTest.*", "**/WorkerSinkTaskTest.*", "**/WorkerSinkTaskThreadedTest.*", - "**/WorkerSourceTaskTest.*", "**/AbstractWorkerSourceTaskTest.*" + "**/WorkerSourceTaskTest.*" ]) } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java index f2f63264e3653..858921b539ed9 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java @@ -16,20 +16,21 @@ */ package org.apache.kafka.connect.runtime; -import org.apache.kafka.clients.admin.NewTopic; +import java.util.stream.Collectors; import org.apache.kafka.clients.admin.TopicDescription; +import java.util.Arrays; import org.apache.kafka.clients.producer.Callback; +import org.apache.kafka.clients.admin.NewTopic; 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.InvalidRecordException; import org.apache.kafka.common.MetricName; -import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.TopicPartitionInfo; import org.apache.kafka.common.errors.TopicAuthorizationException; -import org.apache.kafka.common.header.Header; -import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.header.internals.RecordHeaders; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.utils.Time; import org.apache.kafka.connect.data.Schema; import org.apache.kafka.connect.data.SchemaAndValue; @@ -52,31 +53,26 @@ import org.apache.kafka.connect.storage.StringConverter; import org.apache.kafka.connect.util.ConnectorTaskId; import org.apache.kafka.connect.util.TopicAdmin; +import org.apache.kafka.connect.util.TopicAdmin.TopicCreationResponse; import org.apache.kafka.connect.util.TopicCreationGroup; -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.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.junit.MockitoJUnitRunner; +import org.mockito.stubbing.Answer; import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.Arrays; +import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; -import java.util.concurrent.Future; import java.util.concurrent.TimeoutException; import static org.apache.kafka.connect.integration.MonitorableSourceConnector.TOPIC_CONFIG; @@ -94,12 +90,17 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; import static org.junit.Assert.assertThrows; -import static org.junit.Assert.assertTrue; -@PowerMockIgnore({"javax.management.*", - "org.apache.log4j.*"}) -@RunWith(PowerMockRunner.class) +@SuppressWarnings("unchecked") +@RunWith(MockitoJUnitRunner.StrictStubs.class) public class AbstractWorkerSourceTaskTest { private static final String TOPIC = "topic"; @@ -117,7 +118,8 @@ public class AbstractWorkerSourceTaskTest { private static final byte[] SERIALIZED_KEY = "converted-key".getBytes(); private static final byte[] SERIALIZED_RECORD = "converted-record".getBytes(); - @Mock private SourceTask sourceTask; + @Mock + private SourceTask sourceTask; @Mock private TopicAdmin admin; @Mock private KafkaProducer producer; @Mock private Converter keyConverter; @@ -129,7 +131,7 @@ public class AbstractWorkerSourceTaskTest { @Mock private ConnectorOffsetBackingStore offsetStore; @Mock private StatusBackingStore statusBackingStore; @Mock private WorkerSourceTaskContext sourceTaskContext; - @MockStrict private TaskStatus.Listener statusListener; + @Mock private TaskStatus.Listener statusListener; private final ConnectorTaskId taskId = new ConnectorTaskId("job", 0); private final ConnectorTaskId taskId1 = new ConnectorTaskId("job", 1); @@ -139,7 +141,6 @@ public class AbstractWorkerSourceTaskTest { private SourceConnectorConfig sourceConfig; private MockConnectMetrics metrics = new MockConnectMetrics(); @Mock private ErrorHandlingMetrics errorHandlingMetrics; - private Capture producerCallbacks; private AbstractWorkerSourceTask workerTask; @@ -148,8 +149,7 @@ public void setup() { Map workerProps = workerProps(); plugins = new Plugins(workerProps); config = new StandaloneConfig(workerProps); - sourceConfig = new SourceConnectorConfig(plugins, sourceConnectorPropsWithGroups(TOPIC), true); - producerCallbacks = EasyMock.newCapture(); + sourceConfig = new SourceConnectorConfig(plugins, sourceConnectorPropsWithGroups(), true); metrics = new MockConnectMetrics(); } @@ -162,21 +162,21 @@ private Map workerProps() { return props; } - private Map sourceConnectorPropsWithGroups(String topic) { + private Map sourceConnectorPropsWithGroups() { // setup up props for the source connector Map props = new HashMap<>(); props.put("name", "foo-connector"); props.put(CONNECTOR_CLASS_CONFIG, MonitorableSourceConnector.class.getSimpleName()); props.put(TASKS_MAX_CONFIG, String.valueOf(1)); - props.put(TOPIC_CONFIG, topic); + props.put(TOPIC_CONFIG, TOPIC); props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName()); props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName()); props.put(TOPIC_CREATION_GROUPS_CONFIG, String.join(",", "foo", "bar")); props.put(DEFAULT_TOPIC_CREATION_PREFIX + REPLICATION_FACTOR_CONFIG, String.valueOf(1)); props.put(DEFAULT_TOPIC_CREATION_PREFIX + PARTITIONS_CONFIG, String.valueOf(1)); - props.put(SourceConnectorConfig.TOPIC_CREATION_PREFIX + "foo" + "." + INCLUDE_REGEX_CONFIG, topic); + props.put(SourceConnectorConfig.TOPIC_CREATION_PREFIX + "foo" + "." + INCLUDE_REGEX_CONFIG, TOPIC); props.put(SourceConnectorConfig.TOPIC_CREATION_PREFIX + "bar" + "." + INCLUDE_REGEX_CONFIG, ".*"); - props.put(SourceConnectorConfig.TOPIC_CREATION_PREFIX + "bar" + "." + EXCLUDE_REGEX_CONFIG, topic); + props.put(SourceConnectorConfig.TOPIC_CREATION_PREFIX + "bar" + "." + EXCLUDE_REGEX_CONFIG, TOPIC); return props; } @@ -235,45 +235,76 @@ public void testMetricsGroup() { public void testSendRecordsConvertsData() { createWorkerTask(); - List records = new ArrayList<>(); // Can just use the same record for key and value - records.add(new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD)); + List records = Collections.singletonList( + new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD) + ); - Capture> sent = expectSendRecordAnyTimes(); + when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); + when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC)); + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + when(statusBackingStore.getTopic(anyString(), anyString())).thenAnswer((Answer) 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()); + }); + 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); + workerTask.toSend = records; + workerTask.sendRecords(); - expectTopicCreation(TOPIC); + ArgumentCaptor> sent = ArgumentCaptor.forClass(ProducerRecord.class); + ArgumentCaptor producerCallbacks = ArgumentCaptor.forClass(Callback.class); + verify(producer).send(sent.capture(), producerCallbacks.capture()); - PowerMock.replayAll(); + for (Callback cb : producerCallbacks.getAllValues()) { + cb.onCompletion(new RecordMetadata(new TopicPartition("foo", 0), 0, 0, 0L, 0, 0), + null); + } - workerTask.toSend = records; - workerTask.sendRecords(); assertEquals(SERIALIZED_KEY, sent.getValue().key()); assertEquals(SERIALIZED_RECORD, sent.getValue().value()); - PowerMock.verifyAll(); + ArgumentCaptor recordCapture = ArgumentCaptor.forClass(SourceRecord.class); + ArgumentCaptor connectorCapture = ArgumentCaptor.forClass(String.class); + ArgumentCaptor topicCapture = ArgumentCaptor.forClass(String.class); + ArgumentCaptor newTopicCapture = ArgumentCaptor.forClass(NewTopic.class); + + verify(transformationChain).apply(recordCapture.capture()); + verify(statusBackingStore).getTopic(connectorCapture.capture(), topicCapture.capture()); + + assertEquals("job", connectorCapture.getValue()); + assertEquals(TOPIC, topicCapture.getValue()); + + verify(admin).createOrFindTopics(newTopicCapture.capture()); + assertEquals(TOPIC, newTopicCapture.getValue().name()); } @Test public void testSendRecordsPropagatesTimestamp() { final Long timestamp = System.currentTimeMillis(); - createWorkerTask(); - List records = Collections.singletonList( + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + 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); + when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC)); + + workerTask.toSend = Collections.singletonList( new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD, timestamp) ); + workerTask.sendRecords(); - Capture> sent = expectSendRecordAnyTimes(); - - expectTopicCreation(TOPIC); - - PowerMock.replayAll(); + ArgumentCaptor> sent = ArgumentCaptor.forClass(ProducerRecord.class); + verify(producer).send(sent.capture(), any()); - workerTask.toSend = records; - workerTask.sendRecords(); assertEquals(timestamp, sent.getValue().timestamp()); - - PowerMock.verifyAll(); } @Test @@ -281,19 +312,18 @@ public void testSendRecordsCorruptTimestamp() { final Long timestamp = -3L; createWorkerTask(); - List records = Collections.singletonList( + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + 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); + + workerTask.toSend = Collections.singletonList( new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD, timestamp) ); - - Capture> sent = expectSendRecordAnyTimes(); - - PowerMock.replayAll(); - - workerTask.toSend = records; assertThrows(InvalidRecordException.class, workerTask::sendRecords); - assertFalse(sent.hasCaptured()); - - PowerMock.verifyAll(); + verifyNoInteractions(producer); } @Test @@ -301,21 +331,23 @@ public void testSendRecordsNoTimestamp() { final Long timestamp = -1L; createWorkerTask(); - List records = Collections.singletonList( + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + 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); + when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC)); + + workerTask.toSend = Collections.singletonList( new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD, timestamp) ); + workerTask.sendRecords(); - Capture> sent = expectSendRecordAnyTimes(); - - expectTopicCreation(TOPIC); - - PowerMock.replayAll(); + ArgumentCaptor> sent = ArgumentCaptor.forClass(ProducerRecord.class); + verify(producer).send(sent.capture(), any()); - workerTask.toSend = records; - workerTask.sendRecords(); assertNull(sent.getValue().timestamp()); - - PowerMock.verifyAll(); } @Test @@ -328,22 +360,34 @@ public void testHeaders() { createWorkerTask(); - List records = new ArrayList<>(); - records.add(new SourceRecord(PARTITION, OFFSET, TOPIC, null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD, null, connectHeaders)); - - expectTopicCreation(TOPIC); - - Capture> sent = expectSendRecord(TOPIC, true, headers); + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), + anyString())) + .thenAnswer((Answer) invocation -> { + String headerValue = invocation.getArgument(3, String.class); + return headerValue.getBytes(StandardCharsets.UTF_8); + }); + 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); + when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC)); + + workerTask.toSend = Collections.singletonList( + new SourceRecord(PARTITION, OFFSET, TOPIC, null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD, + null, connectHeaders) + ); + workerTask.sendRecords(); - PowerMock.replayAll(); + ArgumentCaptor> sent = ArgumentCaptor.forClass( + ProducerRecord.class); + verify(producer).send(sent.capture(), any()); - workerTask.toSend = records; - workerTask.sendRecords(); assertEquals(SERIALIZED_KEY, sent.getValue().key()); assertEquals(SERIALIZED_RECORD, sent.getValue().value()); assertEquals(headers, sent.getValue().headers()); - - PowerMock.verifyAll(); } @Test @@ -353,47 +397,62 @@ public void testHeadersWithCustomConverter() throws Exception { createWorkerTask(stringConverter, testConverter, stringConverter); - List records = new ArrayList<>(); + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), + anyString())) + .thenAnswer((Answer) invocation -> { + String headerValue = invocation.getArgument(3, String.class); + return headerValue.getBytes(StandardCharsets.UTF_8); + }); + 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); + when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC)); String stringA = "Árvíztűrő tükörfúrógép"; - org.apache.kafka.connect.header.Headers headersA = new ConnectHeaders(); String encodingA = "latin2"; - headersA.addString("encoding", encodingA); - - records.add(new SourceRecord(PARTITION, OFFSET, "topic", null, Schema.STRING_SCHEMA, "a", Schema.STRING_SCHEMA, stringA, null, headersA)); - String stringB = "Тестовое сообщение"; - org.apache.kafka.connect.header.Headers headersB = new ConnectHeaders(); String encodingB = "koi8_r"; - headersB.addString("encoding", encodingB); - records.add(new SourceRecord(PARTITION, OFFSET, "topic", null, Schema.STRING_SCHEMA, "b", Schema.STRING_SCHEMA, stringB, null, headersB)); + org.apache.kafka.connect.header.Headers headersA = new ConnectHeaders() + .addString("encoding", encodingA); + org.apache.kafka.connect.header.Headers headersB = new ConnectHeaders() + .addString("encoding", encodingB); - expectTopicCreation(TOPIC); + workerTask.toSend = Arrays.asList( + new SourceRecord(PARTITION, OFFSET, "topic", null, Schema.STRING_SCHEMA, "a", + Schema.STRING_SCHEMA, stringA, null, headersA), + new SourceRecord(PARTITION, OFFSET, "topic", null, Schema.STRING_SCHEMA, "b", + Schema.STRING_SCHEMA, stringB, null, headersB) + ); + workerTask.sendRecords(); - Capture> sentRecordA = expectSendRecord(TOPIC, false, null); - Capture> sentRecordB = expectSendRecord(TOPIC, false, null); + ArgumentCaptor> sent = ArgumentCaptor.forClass( + ProducerRecord.class); + verify(producer, times(2)).send(sent.capture(), any()); - PowerMock.replayAll(); + List> capturedValues = sent.getAllValues(); + assertEquals(2, capturedValues.size()); - workerTask.toSend = records; - workerTask.sendRecords(); + ProducerRecord sentRecordA = capturedValues.get(0); + ProducerRecord sentRecordB = capturedValues.get(1); - assertEquals(ByteBuffer.wrap("a".getBytes()), ByteBuffer.wrap(sentRecordA.getValue().key())); + assertEquals(ByteBuffer.wrap("a".getBytes()), ByteBuffer.wrap(sentRecordA.key())); assertEquals( - ByteBuffer.wrap(stringA.getBytes(encodingA)), - ByteBuffer.wrap(sentRecordA.getValue().value()) + ByteBuffer.wrap(stringA.getBytes(encodingA)), + ByteBuffer.wrap(sentRecordA.value()) ); - assertEquals(encodingA, new String(sentRecordA.getValue().headers().lastHeader("encoding").value())); + assertEquals(encodingA, new String(sentRecordA.headers().lastHeader("encoding").value())); - assertEquals(ByteBuffer.wrap("b".getBytes()), ByteBuffer.wrap(sentRecordB.getValue().key())); + assertEquals(ByteBuffer.wrap("b".getBytes()), ByteBuffer.wrap(sentRecordB.key())); assertEquals( - ByteBuffer.wrap(stringB.getBytes(encodingB)), - ByteBuffer.wrap(sentRecordB.getValue().value()) + ByteBuffer.wrap(stringB.getBytes(encodingB)), + ByteBuffer.wrap(sentRecordB.value()) ); - assertEquals(encodingB, new String(sentRecordB.getValue().headers().lastHeader("encoding").value())); - - PowerMock.verifyAll(); + assertEquals(encodingB, new String(sentRecordB.headers().lastHeader("encoding").value())); } @Test @@ -403,18 +462,30 @@ public void testTopicCreateWhenTopicExists() { 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); - expectPreliminaryCalls(); TopicPartitionInfo topicPartitionInfo = new TopicPartitionInfo(0, null, Collections.emptyList(), Collections.emptyList()); TopicDescription topicDesc = new TopicDescription(TOPIC, false, Collections.singletonList(topicPartitionInfo)); - EasyMock.expect(admin.describeTopics(TOPIC)).andReturn(Collections.singletonMap(TOPIC, topicDesc)); - expectSendRecord(); - expectSendRecord(); - - PowerMock.replayAll(); + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), + anyString())) + .thenAnswer((Answer) invocation -> { + String headerValue = invocation.getArgument(3, String.class); + return headerValue.getBytes(StandardCharsets.UTF_8); + }); + 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); + when(admin.describeTopics(TOPIC)).thenReturn(Collections.singletonMap(TOPIC, topicDesc)); workerTask.toSend = Arrays.asList(record1, record2); workerTask.sendRecords(); + + ArgumentCaptor> sent = ArgumentCaptor.forClass( + ProducerRecord.class); + verify(producer, times(2)).send(sent.capture(), any()); } @Test @@ -424,18 +495,33 @@ public void testSendRecordsTopicDescribeRetries() { 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); - expectPreliminaryCalls(); - // First round - call to describe the topic times out - EasyMock.expect(admin.describeTopics(TOPIC)) - .andThrow(new RetriableException(new TimeoutException("timeout"))); - - // Second round - calls to describe and create succeed - expectTopicCreation(TOPIC); - // Exactly two records are sent - expectSendRecord(); - expectSendRecord(); - - PowerMock.replayAll(); + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), + anyString())) + .thenAnswer((Answer) invocation -> { + String headerValue = invocation.getArgument(3, String.class); + return headerValue.getBytes(StandardCharsets.UTF_8); + }); + 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); + when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC)); + when(admin.describeTopics(TOPIC)) + .thenAnswer(new Answer>() { + boolean firstCall = true; + + @Override + public Map answer(InvocationOnMock invocation) { + if (firstCall) { + firstCall = false; + throw new RetriableException(new TimeoutException("timeout")); + } + return Collections.emptyMap(); + } + }); workerTask.toSend = Arrays.asList(record1, record2); workerTask.sendRecords(); @@ -453,19 +539,33 @@ public void testSendRecordsTopicCreateRetries() { 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); - // First call to describe the topic times out - expectPreliminaryCalls(); - EasyMock.expect(admin.describeTopics(TOPIC)).andReturn(Collections.emptyMap()); - Capture newTopicCapture = EasyMock.newCapture(); - EasyMock.expect(admin.createOrFindTopics(EasyMock.capture(newTopicCapture))) - .andThrow(new RetriableException(new TimeoutException("timeout"))); - - // Second round - expectTopicCreation(TOPIC); - expectSendRecord(); - expectSendRecord(); - - PowerMock.replayAll(); + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), + anyString())) + .thenAnswer((Answer) invocation -> { + String headerValue = invocation.getArgument(3, String.class); + return headerValue.getBytes(StandardCharsets.UTF_8); + }); + 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); + when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); + when(admin.createOrFindTopics(any(NewTopic.class))) + .thenAnswer(new Answer() { + boolean firstCall = true; + + @Override + public TopicAdmin.TopicCreationResponse answer(InvocationOnMock invocation) { + if (firstCall) { + firstCall = false; + throw new RetriableException(new TimeoutException("timeout")); + } + return createdTopic(TOPIC); + } + }); workerTask.toSend = Arrays.asList(record1, record2); workerTask.sendRecords(); @@ -485,32 +585,56 @@ public void testSendRecordsTopicDescribeRetriesMidway() { SourceRecord record2 = new SourceRecord(PARTITION, OFFSET, TOPIC, 2, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD); SourceRecord record3 = new SourceRecord(PARTITION, OFFSET, OTHER_TOPIC, 3, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD); - // First round - expectPreliminaryCalls(OTHER_TOPIC); - expectTopicCreation(TOPIC); - expectSendRecord(); - expectSendRecord(); + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), + anyString())) + .thenAnswer((Answer) invocation -> { + String headerValue = invocation.getArgument(3, String.class); + return headerValue.getBytes(StandardCharsets.UTF_8); + }); + 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); + when(admin.describeTopics(anyString())).thenAnswer(new Answer>() { + int counter = 0; - // First call to describe the topic times out - EasyMock.expect(admin.describeTopics(OTHER_TOPIC)) - .andThrow(new RetriableException(new TimeoutException("timeout"))); - - // Second round - expectTopicCreation(OTHER_TOPIC); - expectSendRecord(OTHER_TOPIC, false, emptyHeaders()); - - PowerMock.replayAll(); + @Override + public Map answer(InvocationOnMock invocation) { + counter++; + if (counter == 2) { + throw new RetriableException(new TimeoutException("timeout")); + } - // Try to send 3, make first pass, second fail. Should save last two + return Collections.emptyMap(); + } + }); + when(admin.createOrFindTopics(any(NewTopic.class))).thenAnswer( + (Answer) invocation -> { + NewTopic newTopic = invocation.getArgument(0); + return createdTopic(newTopic.name()); + }); + + // Try to send 3, make first pass, second fail. Should save last record workerTask.toSend = Arrays.asList(record1, record2, record3); workerTask.sendRecords(); - assertEquals(Arrays.asList(record3), workerTask.toSend); + assertEquals(Collections.singletonList(record3), workerTask.toSend); // Next they all succeed workerTask.sendRecords(); assertNull(workerTask.toSend); - PowerMock.verifyAll(); + verify(admin, times(3)).describeTopics(anyString()); + + ArgumentCaptor newTopicCaptor = ArgumentCaptor.forClass(NewTopic.class); + verify(admin, times(2)).createOrFindTopics(newTopicCaptor.capture()); + + assertEquals(Arrays.asList(TOPIC, OTHER_TOPIC), newTopicCaptor.getAllValues() + .stream() + .map(NewTopic::name) + .collect(Collectors.toList())); } @Test @@ -522,34 +646,51 @@ public void testSendRecordsTopicCreateRetriesMidway() { SourceRecord record2 = new SourceRecord(PARTITION, OFFSET, TOPIC, 2, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD); SourceRecord record3 = new SourceRecord(PARTITION, OFFSET, OTHER_TOPIC, 3, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD); - // First round - expectPreliminaryCalls(OTHER_TOPIC); - expectTopicCreation(TOPIC); - expectSendRecord(); - expectSendRecord(); + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), + anyString())) + .thenAnswer((Answer) invocation -> { + String headerValue = invocation.getArgument(3, String.class); + return headerValue.getBytes(StandardCharsets.UTF_8); + }); + 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); + when(admin.describeTopics(anyString())).thenReturn(Collections.emptyMap()); + when(admin.createOrFindTopics(any(NewTopic.class))).thenAnswer(new Answer() { + int counter = 0; - EasyMock.expect(admin.describeTopics(OTHER_TOPIC)).andReturn(Collections.emptyMap()); - // First call to create the topic times out - Capture newTopicCapture = EasyMock.newCapture(); - EasyMock.expect(admin.createOrFindTopics(EasyMock.capture(newTopicCapture))) - .andThrow(new RetriableException(new TimeoutException("timeout"))); - - // Second round - expectTopicCreation(OTHER_TOPIC); - expectSendRecord(OTHER_TOPIC, false, emptyHeaders()); - - PowerMock.replayAll(); + @Override + public TopicAdmin.TopicCreationResponse answer(InvocationOnMock invocation) { + counter++; + if (counter == 2) { + throw new RetriableException(new TimeoutException("timeout")); + } + NewTopic newTopic = invocation.getArgument(0); + return createdTopic(newTopic.name()); + } + }); // Try to send 3, make first pass, second fail. Should save last two workerTask.toSend = Arrays.asList(record1, record2, record3); workerTask.sendRecords(); - assertEquals(Arrays.asList(record3), workerTask.toSend); + assertEquals(Collections.singletonList(record3), workerTask.toSend); // Next they all succeed workerTask.sendRecords(); assertNull(workerTask.toSend); - PowerMock.verifyAll(); + verify(admin, times(3)).describeTopics(anyString()); + ArgumentCaptor newTopicCaptor = ArgumentCaptor.forClass(NewTopic.class); + verify(admin, times(3)).createOrFindTopics(newTopicCaptor.capture()); + + assertEquals(Arrays.asList(TOPIC, OTHER_TOPIC, OTHER_TOPIC), newTopicCaptor.getAllValues() + .stream() + .map(NewTopic::name) + .collect(Collectors.toList())); } @Test @@ -559,11 +700,22 @@ public void testTopicDescribeFails() { 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); - expectPreliminaryCalls(); - EasyMock.expect(admin.describeTopics(TOPIC)) - .andThrow(new ConnectException(new TopicAuthorizationException("unauthorized"))); - - PowerMock.replayAll(); + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), + anyString())) + .thenAnswer((Answer) invocation -> { + String headerValue = invocation.getArgument(3, String.class); + return headerValue.getBytes(StandardCharsets.UTF_8); + }); + 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); + when(admin.describeTopics(TOPIC)).thenThrow( + new ConnectException(new TopicAuthorizationException("unauthorized")) + ); workerTask.toSend = Arrays.asList(record1, record2); assertThrows(ConnectException.class, workerTask::sendRecords); @@ -576,18 +728,27 @@ public void testTopicCreateFails() { 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); - expectPreliminaryCalls(); - EasyMock.expect(admin.describeTopics(TOPIC)).andReturn(Collections.emptyMap()); - - Capture newTopicCapture = EasyMock.newCapture(); - EasyMock.expect(admin.createOrFindTopics(EasyMock.capture(newTopicCapture))) - .andThrow(new ConnectException(new TopicAuthorizationException("unauthorized"))); - - PowerMock.replayAll(); + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), + anyString())) + .thenAnswer((Answer) invocation -> { + String headerValue = invocation.getArgument(3, String.class); + return headerValue.getBytes(StandardCharsets.UTF_8); + }); + 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); + when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); + when(admin.createOrFindTopics(any(NewTopic.class))).thenThrow( + new ConnectException(new TopicAuthorizationException("unauthorized")) + ); workerTask.toSend = Arrays.asList(record1, record2); assertThrows(ConnectException.class, workerTask::sendRecords); - assertTrue(newTopicCapture.hasCaptured()); + verify(admin).createOrFindTopics(any()); } @Test @@ -597,17 +758,25 @@ public void testTopicCreateFailsWithExceptionWhenCreateReturnsTopicNotCreatedOrF 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); - expectPreliminaryCalls(); - EasyMock.expect(admin.describeTopics(TOPIC)).andReturn(Collections.emptyMap()); - - Capture newTopicCapture = EasyMock.newCapture(); - EasyMock.expect(admin.createOrFindTopics(EasyMock.capture(newTopicCapture))).andReturn(TopicAdmin.EMPTY_CREATION); - - PowerMock.replayAll(); + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), + anyString())) + .thenAnswer((Answer) invocation -> { + String headerValue = invocation.getArgument(3, String.class); + return headerValue.getBytes(StandardCharsets.UTF_8); + }); + 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); + when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); + when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(TopicAdmin.EMPTY_CREATION); workerTask.toSend = Arrays.asList(record1, record2); assertThrows(ConnectException.class, workerTask::sendRecords); - assertTrue(newTopicCapture.hasCaptured()); + verify(admin).createOrFindTopics(any()); } @Test @@ -617,19 +786,25 @@ public void testTopicCreateSucceedsWhenCreateReturnsExistingTopicFound() { 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); - expectPreliminaryCalls(); - EasyMock.expect(admin.describeTopics(TOPIC)).andReturn(Collections.emptyMap()); - - Capture newTopicCapture = EasyMock.newCapture(); - EasyMock.expect(admin.createOrFindTopics(EasyMock.capture(newTopicCapture))).andReturn(foundTopic(TOPIC)); - - expectSendRecord(); - expectSendRecord(); - - PowerMock.replayAll(); + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), + anyString())) + .thenAnswer((Answer) invocation -> { + String headerValue = invocation.getArgument(3, String.class); + return headerValue.getBytes(StandardCharsets.UTF_8); + }); + 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); + when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); + when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(foundTopic(TOPIC)); workerTask.toSend = Arrays.asList(record1, record2); workerTask.sendRecords(); + verify(producer, times(2)).send(any(ProducerRecord.class), any(Callback.class)); } @Test @@ -639,93 +814,25 @@ public void testTopicCreateSucceedsWhenCreateReturnsNewTopicFound() { 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); - expectPreliminaryCalls(); - EasyMock.expect(admin.describeTopics(TOPIC)).andReturn(Collections.emptyMap()); - - Capture newTopicCapture = EasyMock.newCapture(); - EasyMock.expect(admin.createOrFindTopics(EasyMock.capture(newTopicCapture))).andReturn(createdTopic(TOPIC)); - - expectSendRecord(); - expectSendRecord(); - - PowerMock.replayAll(); + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), + anyString())) + .thenAnswer((Answer) invocation -> { + String headerValue = invocation.getArgument(3, String.class); + return headerValue.getBytes(StandardCharsets.UTF_8); + }); + 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); + when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); + when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC)); workerTask.toSend = Arrays.asList(record1, record2); workerTask.sendRecords(); - } - - private Capture> expectSendRecord( - String topic, - boolean anyTimes, - Headers headers - ) { - if (headers != null) - expectConvertHeadersAndKeyValue(topic, anyTimes, headers); - - expectApplyTransformationChain(anyTimes); - - Capture> sent = EasyMock.newCapture(); - - IExpectationSetters> expect = EasyMock.expect( - producer.send(EasyMock.capture(sent), EasyMock.capture(producerCallbacks))); - - IAnswer> expectResponse = () -> { - synchronized (producerCallbacks) { - for (Callback cb : producerCallbacks.getValues()) { - cb.onCompletion(new RecordMetadata(new TopicPartition("foo", 0), 0, 0, 0L, 0, 0), null); - } - producerCallbacks.reset(); - } - return null; - }; - - if (anyTimes) - expect.andStubAnswer(expectResponse); - else - expect.andAnswer(expectResponse); - - expectTaskGetTopic(anyTimes); - - return sent; - } - - private Capture> expectSendRecordAnyTimes() { - return expectSendRecord(TOPIC, true, emptyHeaders()); - } - - private Capture> expectSendRecord() { - return expectSendRecord(TOPIC, false, emptyHeaders()); - } - - private void expectTaskGetTopic(boolean anyTimes) { - final Capture connectorCapture = EasyMock.newCapture(); - final Capture topicCapture = EasyMock.newCapture(); - IExpectationSetters 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 expectTopicCreation(String topic) { - if (config.topicCreationEnable()) { - EasyMock.expect(admin.describeTopics(topic)).andReturn(Collections.emptyMap()); - Capture newTopicCapture = EasyMock.newCapture(); - EasyMock.expect(admin.createOrFindTopics(EasyMock.capture(newTopicCapture))).andReturn(createdTopic(topic)); - } + verify(producer, times(2)).send(any(ProducerRecord.class), any(Callback.class)); } private TopicAdmin.TopicCreationResponse createdTopic(String topic) { @@ -740,49 +847,6 @@ private TopicAdmin.TopicCreationResponse foundTopic(String topic) { return new TopicAdmin.TopicCreationResponse(created, existing); } - private void expectPreliminaryCalls() { - expectPreliminaryCalls(TOPIC); - } - - private void expectPreliminaryCalls(String topic) { - expectConvertHeadersAndKeyValue(topic, true, emptyHeaders()); - expectApplyTransformationChain(false); - PowerMock.expectLastCall(); - } - - private void expectConvertHeadersAndKeyValue(String topic, boolean anyTimes, Headers headers) { - for (Header header : headers) { - IExpectationSetters 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 convertKeyExpect = EasyMock.expect(keyConverter.fromConnectData(topic, headers, KEY_SCHEMA, KEY)); - if (anyTimes) - convertKeyExpect.andStubReturn(SERIALIZED_KEY); - else - convertKeyExpect.andReturn(SERIALIZED_KEY); - IExpectationSetters convertValueExpect = EasyMock.expect(valueConverter.fromConnectData(topic, headers, RECORD_SCHEMA, RECORD)); - if (anyTimes) - convertValueExpect.andStubReturn(SERIALIZED_RECORD); - else - convertValueExpect.andReturn(SERIALIZED_RECORD); - } - - private void expectApplyTransformationChain(boolean anyTimes) { - final Capture recordCapture = EasyMock.newCapture(); - IExpectationSetters convertKeyExpect = EasyMock.expect(transformationChain.apply(EasyMock.capture(recordCapture))); - if (anyTimes) - convertKeyExpect.andStubAnswer(recordCapture::getValue); - else - convertKeyExpect.andAnswer(recordCapture::getValue); - } - - private RecordHeaders emptyHeaders() { - return new RecordHeaders(); - } - private void createWorkerTask() { createWorkerTask(keyConverter, valueConverter, headerConverter); } @@ -838,7 +902,5 @@ protected void producerSendFailed(boolean synchronous, ProducerRecord Date: Mon, 6 Feb 2023 22:25:41 -0500 Subject: [PATCH 2/7] PR feedback --- .../runtime/AbstractWorkerSourceTaskTest.java | 383 +++++++----------- 1 file changed, 146 insertions(+), 237 deletions(-) diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java index 858921b539ed9..03ef9feebada7 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java @@ -53,7 +53,6 @@ import org.apache.kafka.connect.storage.StringConverter; import org.apache.kafka.connect.util.ConnectorTaskId; import org.apache.kafka.connect.util.TopicAdmin; -import org.apache.kafka.connect.util.TopicAdmin.TopicCreationResponse; import org.apache.kafka.connect.util.TopicCreationGroup; import org.junit.After; import org.junit.Before; @@ -96,6 +95,7 @@ import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; import static org.junit.Assert.assertThrows; @@ -183,6 +183,7 @@ private Map sourceConnectorPropsWithGroups() { @After public void tearDown() { if (metrics != null) metrics.stop(); + verifyNoMoreInteractions(statusListener); } @Test @@ -240,47 +241,18 @@ public void testSendRecordsConvertsData() { new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD) ); - when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); - when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC)); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - when(statusBackingStore.getTopic(anyString(), anyString())).thenAnswer((Answer) 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()); - }); - 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); + expectSendRecord(emptyHeaders()); + expectTopicCreation(TOPIC); + workerTask.toSend = records; workerTask.sendRecords(); - ArgumentCaptor> sent = ArgumentCaptor.forClass(ProducerRecord.class); - ArgumentCaptor producerCallbacks = ArgumentCaptor.forClass(Callback.class); - verify(producer).send(sent.capture(), producerCallbacks.capture()); - - for (Callback cb : producerCallbacks.getAllValues()) { - cb.onCompletion(new RecordMetadata(new TopicPartition("foo", 0), 0, 0, 0L, 0, 0), - null); - } + ArgumentCaptor> sent = verifySendRecord(); assertEquals(SERIALIZED_KEY, sent.getValue().key()); assertEquals(SERIALIZED_RECORD, sent.getValue().value()); - ArgumentCaptor recordCapture = ArgumentCaptor.forClass(SourceRecord.class); - ArgumentCaptor connectorCapture = ArgumentCaptor.forClass(String.class); - ArgumentCaptor topicCapture = ArgumentCaptor.forClass(String.class); - ArgumentCaptor newTopicCapture = ArgumentCaptor.forClass(NewTopic.class); - - verify(transformationChain).apply(recordCapture.capture()); - verify(statusBackingStore).getTopic(connectorCapture.capture(), topicCapture.capture()); - - assertEquals("job", connectorCapture.getValue()); - assertEquals(TOPIC, topicCapture.getValue()); - - verify(admin).createOrFindTopics(newTopicCapture.capture()); - assertEquals(TOPIC, newTopicCapture.getValue().name()); + verifyTaskGetTopic(); } @Test @@ -288,23 +260,18 @@ public void testSendRecordsPropagatesTimestamp() { final Long timestamp = System.currentTimeMillis(); createWorkerTask(); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - 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); - when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC)); + expectSendRecord(emptyHeaders()); + expectTopicCreation(TOPIC); workerTask.toSend = Collections.singletonList( new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD, timestamp) ); workerTask.sendRecords(); - ArgumentCaptor> sent = ArgumentCaptor.forClass(ProducerRecord.class); - verify(producer).send(sent.capture(), any()); - + ArgumentCaptor> sent = verifySendRecord(); assertEquals(timestamp, sent.getValue().timestamp()); + + verifyTaskGetTopic(); } @Test @@ -312,12 +279,8 @@ public void testSendRecordsCorruptTimestamp() { final Long timestamp = -3L; createWorkerTask(); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - 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); + expectSendRecord(emptyHeaders()); + expectTopicCreation(TOPIC); workerTask.toSend = Collections.singletonList( new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD, timestamp) @@ -331,49 +294,30 @@ public void testSendRecordsNoTimestamp() { final Long timestamp = -1L; createWorkerTask(); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - 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); - when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC)); + expectSendRecord(emptyHeaders()); + expectTopicCreation(TOPIC); workerTask.toSend = Collections.singletonList( new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD, timestamp) ); workerTask.sendRecords(); - ArgumentCaptor> sent = ArgumentCaptor.forClass(ProducerRecord.class); - verify(producer).send(sent.capture(), any()); - + ArgumentCaptor> sent = verifySendRecord(); assertNull(sent.getValue().timestamp()); } @Test public void testHeaders() { - Headers headers = new RecordHeaders(); - headers.add("header_key", "header_value".getBytes()); + Headers headers = new RecordHeaders() + .add("header_key", "header_value".getBytes()); - org.apache.kafka.connect.header.Headers connectHeaders = new ConnectHeaders(); - connectHeaders.add("header_key", new SchemaAndValue(Schema.STRING_SCHEMA, "header_value")); + org.apache.kafka.connect.header.Headers connectHeaders = new ConnectHeaders() + .add("header_key", new SchemaAndValue(Schema.STRING_SCHEMA, "header_value")); createWorkerTask(); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), - anyString())) - .thenAnswer((Answer) invocation -> { - String headerValue = invocation.getArgument(3, String.class); - return headerValue.getBytes(StandardCharsets.UTF_8); - }); - 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); - when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC)); + expectSendRecord(headers); + expectTopicCreation(TOPIC); workerTask.toSend = Collections.singletonList( new SourceRecord(PARTITION, OFFSET, TOPIC, null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD, @@ -381,9 +325,7 @@ public void testHeaders() { ); workerTask.sendRecords(); - ArgumentCaptor> sent = ArgumentCaptor.forClass( - ProducerRecord.class); - verify(producer).send(sent.capture(), any()); + ArgumentCaptor> sent = verifySendRecord(); assertEquals(SERIALIZED_KEY, sent.getValue().key()); assertEquals(SERIALIZED_RECORD, sent.getValue().value()); @@ -397,20 +339,8 @@ public void testHeadersWithCustomConverter() throws Exception { createWorkerTask(stringConverter, testConverter, stringConverter); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), - anyString())) - .thenAnswer((Answer) invocation -> { - String headerValue = invocation.getArgument(3, String.class); - return headerValue.getBytes(StandardCharsets.UTF_8); - }); - 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); - when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC)); + expectSendRecord(null); + expectTopicCreation(TOPIC); String stringA = "Árvíztűrő tükörfúrógép"; String encodingA = "latin2"; @@ -430,9 +360,7 @@ public void testHeadersWithCustomConverter() throws Exception { ); workerTask.sendRecords(); - ArgumentCaptor> sent = ArgumentCaptor.forClass( - ProducerRecord.class); - verify(producer, times(2)).send(sent.capture(), any()); + ArgumentCaptor> sent = verifySendRecord(2); List> capturedValues = sent.getAllValues(); assertEquals(2, capturedValues.size()); @@ -462,30 +390,18 @@ public void testTopicCreateWhenTopicExists() { 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); + expectPreliminaryCalls(); + TopicPartitionInfo topicPartitionInfo = new TopicPartitionInfo(0, null, Collections.emptyList(), Collections.emptyList()); TopicDescription topicDesc = new TopicDescription(TOPIC, false, Collections.singletonList(topicPartitionInfo)); - - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), - anyString())) - .thenAnswer((Answer) invocation -> { - String headerValue = invocation.getArgument(3, String.class); - return headerValue.getBytes(StandardCharsets.UTF_8); - }); - 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); when(admin.describeTopics(TOPIC)).thenReturn(Collections.singletonMap(TOPIC, topicDesc)); + expectSendRecord(emptyHeaders()); + workerTask.toSend = Arrays.asList(record1, record2); workerTask.sendRecords(); - ArgumentCaptor> sent = ArgumentCaptor.forClass( - ProducerRecord.class); - verify(producer, times(2)).send(sent.capture(), any()); + verifySendRecord(2); } @Test @@ -495,20 +411,9 @@ public void testSendRecordsTopicDescribeRetries() { 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); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), - anyString())) - .thenAnswer((Answer) invocation -> { - String headerValue = invocation.getArgument(3, String.class); - return headerValue.getBytes(StandardCharsets.UTF_8); - }); - 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); - when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC)); + expectPreliminaryCalls(); + expectTopicCreation(TOPIC); + when(admin.describeTopics(TOPIC)) .thenAnswer(new Answer>() { boolean firstCall = true; @@ -539,19 +444,8 @@ public void testSendRecordsTopicCreateRetries() { 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); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), - anyString())) - .thenAnswer((Answer) invocation -> { - String headerValue = invocation.getArgument(3, String.class); - return headerValue.getBytes(StandardCharsets.UTF_8); - }); - 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); + expectPreliminaryCalls(); + when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); when(admin.createOrFindTopics(any(NewTopic.class))) .thenAnswer(new Answer() { @@ -585,19 +479,8 @@ public void testSendRecordsTopicDescribeRetriesMidway() { SourceRecord record2 = new SourceRecord(PARTITION, OFFSET, TOPIC, 2, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD); SourceRecord record3 = new SourceRecord(PARTITION, OFFSET, OTHER_TOPIC, 3, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), - anyString())) - .thenAnswer((Answer) invocation -> { - String headerValue = invocation.getArgument(3, String.class); - return headerValue.getBytes(StandardCharsets.UTF_8); - }); - 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); + expectPreliminaryCalls(); + when(admin.describeTopics(anyString())).thenAnswer(new Answer>() { int counter = 0; @@ -612,7 +495,7 @@ public Map answer(InvocationOnMock invocation) { } }); when(admin.createOrFindTopics(any(NewTopic.class))).thenAnswer( - (Answer) invocation -> { + (Answer) invocation -> { NewTopic newTopic = invocation.getArgument(0); return createdTopic(newTopic.name()); }); @@ -646,19 +529,8 @@ public void testSendRecordsTopicCreateRetriesMidway() { SourceRecord record2 = new SourceRecord(PARTITION, OFFSET, TOPIC, 2, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD); SourceRecord record3 = new SourceRecord(PARTITION, OFFSET, OTHER_TOPIC, 3, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), - anyString())) - .thenAnswer((Answer) invocation -> { - String headerValue = invocation.getArgument(3, String.class); - return headerValue.getBytes(StandardCharsets.UTF_8); - }); - 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); + expectPreliminaryCalls(); + when(admin.describeTopics(anyString())).thenReturn(Collections.emptyMap()); when(admin.createOrFindTopics(any(NewTopic.class))).thenAnswer(new Answer() { int counter = 0; @@ -700,19 +572,7 @@ public void testTopicDescribeFails() { 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); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), - anyString())) - .thenAnswer((Answer) invocation -> { - String headerValue = invocation.getArgument(3, String.class); - return headerValue.getBytes(StandardCharsets.UTF_8); - }); - 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); + expectPreliminaryCalls(); when(admin.describeTopics(TOPIC)).thenThrow( new ConnectException(new TopicAuthorizationException("unauthorized")) ); @@ -728,19 +588,7 @@ public void testTopicCreateFails() { 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); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), - anyString())) - .thenAnswer((Answer) invocation -> { - String headerValue = invocation.getArgument(3, String.class); - return headerValue.getBytes(StandardCharsets.UTF_8); - }); - 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); + expectPreliminaryCalls(); when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); when(admin.createOrFindTopics(any(NewTopic.class))).thenThrow( new ConnectException(new TopicAuthorizationException("unauthorized")) @@ -758,19 +606,8 @@ public void testTopicCreateFailsWithExceptionWhenCreateReturnsTopicNotCreatedOrF 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); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), - anyString())) - .thenAnswer((Answer) invocation -> { - String headerValue = invocation.getArgument(3, String.class); - return headerValue.getBytes(StandardCharsets.UTF_8); - }); - 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); + expectPreliminaryCalls(); + when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(TopicAdmin.EMPTY_CREATION); @@ -786,25 +623,18 @@ public void testTopicCreateSucceedsWhenCreateReturnsExistingTopicFound() { 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); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), - anyString())) - .thenAnswer((Answer) invocation -> { - String headerValue = invocation.getArgument(3, String.class); - return headerValue.getBytes(StandardCharsets.UTF_8); - }); - 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); + expectSendRecord(emptyHeaders()); + when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(foundTopic(TOPIC)); workerTask.toSend = Arrays.asList(record1, record2); workerTask.sendRecords(); - verify(producer, times(2)).send(any(ProducerRecord.class), any(Callback.class)); + + ArgumentCaptor> sent = verifySendRecord(2); + + List> capturedValues = sent.getAllValues(); + assertEquals(2, capturedValues.size()); } @Test @@ -814,39 +644,118 @@ public void testTopicCreateSucceedsWhenCreateReturnsNewTopicFound() { 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); - when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); - when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), - anyString())) - .thenAnswer((Answer) invocation -> { - String headerValue = invocation.getArgument(3, String.class); - return headerValue.getBytes(StandardCharsets.UTF_8); - }); - 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); + expectSendRecord(emptyHeaders()); + when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(TOPIC)); workerTask.toSend = Arrays.asList(record1, record2); workerTask.sendRecords(); - verify(producer, times(2)).send(any(ProducerRecord.class), any(Callback.class)); + + ArgumentCaptor> sent = verifySendRecord(2); + + List> capturedValues = sent.getAllValues(); + assertEquals(2, capturedValues.size()); + } + + private void expectSendRecord(Headers headers) { + if (headers != null) + expectConvertHeadersAndKeyValue(headers); + + expectApplyTransformationChain(); + + expectTaskGetTopic(); } + private ArgumentCaptor> verifySendRecord() { + return verifySendRecord(1); + } + + private ArgumentCaptor> verifySendRecord(int times) { + ArgumentCaptor> sent = ArgumentCaptor.forClass(ProducerRecord.class); + ArgumentCaptor producerCallbacks = ArgumentCaptor.forClass(Callback.class); + verify(producer, times(times)).send(sent.capture(), producerCallbacks.capture()); + + for (Callback cb : producerCallbacks.getAllValues()) { + cb.onCompletion(new RecordMetadata(new TopicPartition("foo", 0), 0, 0, 0L, 0, 0), + null); + } + + return sent; + } + + private void expectTaskGetTopic() { + when(statusBackingStore.getTopic(anyString(), anyString())).thenAnswer((Answer) 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 verifyTaskGetTopic() { + ArgumentCaptor connectorCapture = ArgumentCaptor.forClass(String.class); + ArgumentCaptor topicCapture = ArgumentCaptor.forClass(String.class); + ArgumentCaptor newTopicCapture = ArgumentCaptor.forClass(NewTopic.class); + verify(statusBackingStore).getTopic(connectorCapture.capture(), topicCapture.capture()); + + assertEquals("job", connectorCapture.getValue()); + assertEquals(TOPIC, topicCapture.getValue()); + + verify(admin).createOrFindTopics(newTopicCapture.capture()); + assertEquals(TOPIC, newTopicCapture.getValue().name()); + } + + @SuppressWarnings("SameParameterValue") + private void expectTopicCreation(String topic) { + when(admin.describeTopics(topic)).thenReturn(Collections.emptyMap()); + when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(topic)); + } + + @SuppressWarnings("SameParameterValue") private TopicAdmin.TopicCreationResponse createdTopic(String topic) { Set created = Collections.singleton(topic); Set existing = Collections.emptySet(); return new TopicAdmin.TopicCreationResponse(created, existing); } + @SuppressWarnings("SameParameterValue") private TopicAdmin.TopicCreationResponse foundTopic(String topic) { Set created = Collections.emptySet(); Set existing = Collections.singleton(topic); return new TopicAdmin.TopicCreationResponse(created, existing); } + private void expectPreliminaryCalls() { + expectConvertHeadersAndKeyValue(emptyHeaders()); + expectApplyTransformationChain(); + } + + private void expectConvertHeadersAndKeyValue(Headers headers) { + if (headers.iterator().hasNext()) { + when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), + anyString())) + .thenAnswer((Answer) invocation -> { + String headerValue = invocation.getArgument(3, String.class); + return headerValue.getBytes(StandardCharsets.UTF_8); + }); + } + + 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 expectApplyTransformationChain() { + when(transformationChain.apply(any(SourceRecord.class))) + .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + } + + private RecordHeaders emptyHeaders() { + return new RecordHeaders(); + } + private void createWorkerTask() { createWorkerTask(keyConverter, valueConverter, headerConverter); } From 642d58ce3bb80c6079ab821c4849480ebcb2dff3 Mon Sep 17 00:00:00 2001 From: Hector Geraldino Date: Mon, 20 Feb 2023 20:54:18 -0500 Subject: [PATCH 3/7] Update connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java Co-authored-by: Chris Egerton --- .../kafka/connect/runtime/AbstractWorkerSourceTaskTest.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java index 03ef9feebada7..d5be4eaddd708 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java @@ -402,6 +402,8 @@ public void testTopicCreateWhenTopicExists() { workerTask.sendRecords(); verifySendRecord(2); + // Make sure we didn't try to create the topic after finding out it already existed + verifyNoMoreInteractions(admin); } @Test From ca2d027946bec2573624b6d32025de300b9bd845 Mon Sep 17 00:00:00 2001 From: Hector Geraldino Date: Mon, 20 Feb 2023 20:58:16 -0500 Subject: [PATCH 4/7] Update connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java Co-authored-by: Chris Egerton --- .../runtime/AbstractWorkerSourceTaskTest.java | 16 ++++------------ 1 file changed, 4 insertions(+), 12 deletions(-) diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java index d5be4eaddd708..923de7d26d9e0 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java @@ -450,18 +450,10 @@ public void testSendRecordsTopicCreateRetries() { when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); when(admin.createOrFindTopics(any(NewTopic.class))) - .thenAnswer(new Answer() { - boolean firstCall = true; - - @Override - public TopicAdmin.TopicCreationResponse answer(InvocationOnMock invocation) { - if (firstCall) { - firstCall = false; - throw new RetriableException(new TimeoutException("timeout")); - } - return createdTopic(TOPIC); - } - }); + // First call to create the topic times out + .thenThrow(new RetriableException(new TimeoutException("timeout"))) + // Next attempt succeeds + .thenReturn(createdTopic(TOPIC)); workerTask.toSend = Arrays.asList(record1, record2); workerTask.sendRecords(); From d465d427fb7fa2657cf9b69ca5e8ad813da12175 Mon Sep 17 00:00:00 2001 From: Hector Geraldino Date: Mon, 20 Feb 2023 21:59:02 -0500 Subject: [PATCH 5/7] PR feedback --- .../runtime/AbstractWorkerSourceTaskTest.java | 120 ++++++++---------- 1 file changed, 51 insertions(+), 69 deletions(-) diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java index 3ebaff434aa31..992335784216d 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java @@ -60,7 +60,6 @@ import org.junit.runner.RunWith; import org.mockito.ArgumentCaptor; import org.mockito.Mock; -import org.mockito.invocation.InvocationOnMock; import org.mockito.junit.MockitoJUnitRunner; import org.mockito.stubbing.Answer; @@ -93,6 +92,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; @@ -254,6 +254,7 @@ public void testSendRecordsConvertsData() { assertArrayEquals(SERIALIZED_RECORD, sent.getValue().value()); verifyTaskGetTopic(); + verifyTopicCreation(); } @Test @@ -273,6 +274,7 @@ public void testSendRecordsPropagatesTimestamp() { assertEquals(timestamp, sent.getValue().timestamp()); verifyTaskGetTopic(); + verifyTopicCreation(); } @Test @@ -281,7 +283,6 @@ public void testSendRecordsCorruptTimestamp() { createWorkerTask(); expectSendRecord(emptyHeaders()); - expectTopicCreation(TOPIC); workerTask.toSend = Collections.singletonList( new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD, timestamp) @@ -391,7 +392,7 @@ public void testTopicCreateWhenTopicExists() { 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); - expectPreliminaryCalls(); + expectPreliminaryCalls(TOPIC); TopicPartitionInfo topicPartitionInfo = new TopicPartitionInfo(0, null, Collections.emptyList(), Collections.emptyList()); TopicDescription topicDesc = new TopicDescription(TOPIC, false, Collections.singletonList(topicPartitionInfo)); @@ -403,6 +404,7 @@ public void testTopicCreateWhenTopicExists() { workerTask.sendRecords(); verifySendRecord(2); + verify(admin, never()).createOrFindTopics(any(NewTopic.class)); // Make sure we didn't try to create the topic after finding out it already existed verifyNoMoreInteractions(admin); } @@ -414,30 +416,24 @@ public void testSendRecordsTopicDescribeRetries() { 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); - expectPreliminaryCalls(); - expectTopicCreation(TOPIC); + expectPreliminaryCalls(TOPIC); when(admin.describeTopics(TOPIC)) - .thenAnswer(new Answer>() { - boolean firstCall = true; - - @Override - public Map answer(InvocationOnMock invocation) { - if (firstCall) { - firstCall = false; - throw new RetriableException(new TimeoutException("timeout")); - } - return Collections.emptyMap(); - } - }); + .thenThrow(new RetriableException(new TimeoutException("timeout"))) + .thenReturn(Collections.emptyMap()); workerTask.toSend = Arrays.asList(record1, record2); workerTask.sendRecords(); assertEquals(Arrays.asList(record1, record2), workerTask.toSend); + verify(admin, never()).createOrFindTopics(any(NewTopic.class)); + verifyNoMoreInteractions(admin); - // Next they all succeed + // Second round - calls to describe and create succeed + expectTopicCreation(TOPIC); workerTask.sendRecords(); assertNull(workerTask.toSend); + + verifyTopicCreation(); } @Test @@ -447,7 +443,7 @@ public void testSendRecordsTopicCreateRetries() { 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); - expectPreliminaryCalls(); + expectPreliminaryCalls(TOPIC); when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); when(admin.createOrFindTopics(any(NewTopic.class))) @@ -474,21 +470,13 @@ public void testSendRecordsTopicDescribeRetriesMidway() { SourceRecord record2 = new SourceRecord(PARTITION, OFFSET, TOPIC, 2, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD); SourceRecord record3 = new SourceRecord(PARTITION, OFFSET, OTHER_TOPIC, 3, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD); - expectPreliminaryCalls(); - - when(admin.describeTopics(anyString())).thenAnswer(new Answer>() { - int counter = 0; + expectPreliminaryCalls(TOPIC); + expectPreliminaryCalls(OTHER_TOPIC); - @Override - public Map answer(InvocationOnMock invocation) { - counter++; - if (counter == 2) { - throw new RetriableException(new TimeoutException("timeout")); - } - - return Collections.emptyMap(); - } - }); + when(admin.describeTopics(anyString())) + .thenReturn(Collections.emptyMap()) + .thenThrow(new RetriableException(new TimeoutException("timeout"))) + .thenReturn(Collections.emptyMap()); when(admin.createOrFindTopics(any(NewTopic.class))).thenAnswer( (Answer) invocation -> { NewTopic newTopic = invocation.getArgument(0); @@ -524,40 +512,26 @@ public void testSendRecordsTopicCreateRetriesMidway() { SourceRecord record2 = new SourceRecord(PARTITION, OFFSET, TOPIC, 2, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD); SourceRecord record3 = new SourceRecord(PARTITION, OFFSET, OTHER_TOPIC, 3, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD); - expectPreliminaryCalls(); + expectPreliminaryCalls(TOPIC); + expectPreliminaryCalls(OTHER_TOPIC); when(admin.describeTopics(anyString())).thenReturn(Collections.emptyMap()); - when(admin.createOrFindTopics(any(NewTopic.class))).thenAnswer(new Answer() { - int counter = 0; - - @Override - public TopicAdmin.TopicCreationResponse answer(InvocationOnMock invocation) { - counter++; - if (counter == 2) { - throw new RetriableException(new TimeoutException("timeout")); - } - NewTopic newTopic = invocation.getArgument(0); - return createdTopic(newTopic.name()); - } - }); + when(admin.createOrFindTopics(any(NewTopic.class))) + .thenReturn(createdTopic(TOPIC)) + .thenThrow(new RetriableException(new TimeoutException("timeout"))) + .thenReturn(createdTopic(OTHER_TOPIC)); // Try to send 3, make first pass, second fail. Should save last two workerTask.toSend = Arrays.asList(record1, record2, record3); workerTask.sendRecords(); assertEquals(Collections.singletonList(record3), workerTask.toSend); + verifyTopicCreation(2, TOPIC, OTHER_TOPIC); // Second call to createOrFindTopics will throw // Next they all succeed workerTask.sendRecords(); assertNull(workerTask.toSend); - verify(admin, times(3)).describeTopics(anyString()); - ArgumentCaptor newTopicCaptor = ArgumentCaptor.forClass(NewTopic.class); - verify(admin, times(3)).createOrFindTopics(newTopicCaptor.capture()); - - assertEquals(Arrays.asList(TOPIC, OTHER_TOPIC, OTHER_TOPIC), newTopicCaptor.getAllValues() - .stream() - .map(NewTopic::name) - .collect(Collectors.toList())); + verifyTopicCreation(3, TOPIC, OTHER_TOPIC, OTHER_TOPIC); } @Test @@ -567,7 +541,7 @@ public void testTopicDescribeFails() { 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); - expectPreliminaryCalls(); + expectPreliminaryCalls(TOPIC); when(admin.describeTopics(TOPIC)).thenThrow( new ConnectException(new TopicAuthorizationException("unauthorized")) ); @@ -583,7 +557,7 @@ public void testTopicCreateFails() { 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); - expectPreliminaryCalls(); + expectPreliminaryCalls(TOPIC); when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); when(admin.createOrFindTopics(any(NewTopic.class))).thenThrow( new ConnectException(new TopicAuthorizationException("unauthorized")) @@ -601,7 +575,7 @@ public void testTopicCreateFailsWithExceptionWhenCreateReturnsTopicNotCreatedOrF 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); - expectPreliminaryCalls(); + expectPreliminaryCalls(TOPIC); when(admin.describeTopics(TOPIC)).thenReturn(Collections.emptyMap()); when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(TopicAdmin.EMPTY_CREATION); @@ -655,7 +629,7 @@ public void testTopicCreateSucceedsWhenCreateReturnsNewTopicFound() { private void expectSendRecord(Headers headers) { if (headers != null) - expectConvertHeadersAndKeyValue(headers); + expectConvertHeadersAndKeyValue(headers, TOPIC); expectApplyTransformationChain(); @@ -690,22 +664,30 @@ private void expectTaskGetTopic() { private void verifyTaskGetTopic() { ArgumentCaptor connectorCapture = ArgumentCaptor.forClass(String.class); ArgumentCaptor topicCapture = ArgumentCaptor.forClass(String.class); - ArgumentCaptor newTopicCapture = ArgumentCaptor.forClass(NewTopic.class); verify(statusBackingStore).getTopic(connectorCapture.capture(), topicCapture.capture()); assertEquals("job", connectorCapture.getValue()); assertEquals(TOPIC, topicCapture.getValue()); - - verify(admin).createOrFindTopics(newTopicCapture.capture()); - assertEquals(TOPIC, newTopicCapture.getValue().name()); } @SuppressWarnings("SameParameterValue") private void expectTopicCreation(String topic) { - when(admin.describeTopics(topic)).thenReturn(Collections.emptyMap()); when(admin.createOrFindTopics(any(NewTopic.class))).thenReturn(createdTopic(topic)); } + private void verifyTopicCreation() { + verifyTopicCreation(1, TOPIC); + } + private void verifyTopicCreation(int times, String... topics) { + ArgumentCaptor newTopicCapture = ArgumentCaptor.forClass(NewTopic.class); + + verify(admin, times(times)).createOrFindTopics(newTopicCapture.capture()); + assertArrayEquals(topics, newTopicCapture.getAllValues() + .stream() + .map(NewTopic::name) + .toArray(String[]::new)); + } + @SuppressWarnings("SameParameterValue") private TopicAdmin.TopicCreationResponse createdTopic(String topic) { Set created = Collections.singleton(topic); @@ -720,12 +702,12 @@ private TopicAdmin.TopicCreationResponse foundTopic(String topic) { return new TopicAdmin.TopicCreationResponse(created, existing); } - private void expectPreliminaryCalls() { - expectConvertHeadersAndKeyValue(emptyHeaders()); + private void expectPreliminaryCalls(String topic) { + expectConvertHeadersAndKeyValue(emptyHeaders(), topic); expectApplyTransformationChain(); } - private void expectConvertHeadersAndKeyValue(Headers headers) { + private void expectConvertHeadersAndKeyValue(Headers headers, String topic) { if (headers.iterator().hasNext()) { when(headerConverter.fromConnectHeader(anyString(), anyString(), eq(Schema.STRING_SCHEMA), anyString())) @@ -735,9 +717,9 @@ private void expectConvertHeadersAndKeyValue(Headers headers) { }); } - when(keyConverter.fromConnectData(eq(TOPIC), any(Headers.class), eq(KEY_SCHEMA), eq(KEY))) + 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), + when(valueConverter.fromConnectData(eq(topic), any(Headers.class), eq(RECORD_SCHEMA), eq(RECORD))) .thenReturn(SERIALIZED_RECORD); } From b8e3dd59ee7501971090d2cf199ae93a051002b0 Mon Sep 17 00:00:00 2001 From: Hector Geraldino Date: Tue, 21 Feb 2023 23:11:00 -0500 Subject: [PATCH 6/7] Use AddtionalAnswers --- .../kafka/connect/runtime/AbstractWorkerSourceTaskTest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java index 992335784216d..2bef8bd6c71f8 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java @@ -58,6 +58,7 @@ import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.AdditionalAnswers; import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; @@ -726,7 +727,7 @@ private void expectConvertHeadersAndKeyValue(Headers headers, String topic) { private void expectApplyTransformationChain() { when(transformationChain.apply(any(SourceRecord.class))) - .thenAnswer((Answer) invocation -> invocation.getArgument(0)); + .thenAnswer(AdditionalAnswers.returnsFirstArg()); } private RecordHeaders emptyHeaders() { From acccb0ee62e470063c06ef0d6f27e6fcbd5f2298 Mon Sep 17 00:00:00 2001 From: Hector Geraldino Date: Sun, 26 Feb 2023 23:31:14 -0500 Subject: [PATCH 7/7] PR feedback --- .../runtime/AbstractWorkerSourceTaskTest.java | 30 +++++++++++++++++-- 1 file changed, 28 insertions(+), 2 deletions(-) diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java index 2bef8bd6c71f8..976c070a5d2c8 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java @@ -290,6 +290,7 @@ public void testSendRecordsCorruptTimestamp() { ); assertThrows(InvalidRecordException.class, workerTask::sendRecords); verifyNoInteractions(producer); + verifyNoInteractions(admin); } @Test @@ -307,6 +308,9 @@ public void testSendRecordsNoTimestamp() { ArgumentCaptor> sent = verifySendRecord(); assertNull(sent.getValue().timestamp()); + + verifyTaskGetTopic(); + verifyTopicCreation(); } @Test @@ -333,6 +337,9 @@ public void testHeaders() { assertArrayEquals(SERIALIZED_KEY, sent.getValue().key()); assertArrayEquals(SERIALIZED_RECORD, sent.getValue().value()); assertEquals(headers, sent.getValue().headers()); + + verifyTaskGetTopic(); + verifyTopicCreation(); } @Test @@ -384,6 +391,9 @@ public void testHeadersWithCustomConverter() throws Exception { ByteBuffer.wrap(sentRecordB.value()) ); assertEquals(encodingB, new String(sentRecordB.headers().lastHeader("encoding").value())); + + verifyTaskGetTopic(2); + verifyTopicCreation(); } @Test @@ -460,6 +470,9 @@ public void testSendRecordsTopicCreateRetries() { // Next they all succeed workerTask.sendRecords(); assertNull(workerTask.toSend); + + // First attempt failed, second succeeded + verifyTopicCreation(2, TOPIC, TOPIC); } @Test @@ -522,7 +535,7 @@ public void testSendRecordsTopicCreateRetriesMidway() { .thenThrow(new RetriableException(new TimeoutException("timeout"))) .thenReturn(createdTopic(OTHER_TOPIC)); - // Try to send 3, make first pass, second fail. Should save last two + // Try to send 3, make first pass, second fail. Should save last record workerTask.toSend = Arrays.asList(record1, record2, record3); workerTask.sendRecords(); assertEquals(Collections.singletonList(record3), workerTask.toSend); @@ -567,6 +580,8 @@ public void testTopicCreateFails() { workerTask.toSend = Arrays.asList(record1, record2); assertThrows(ConnectException.class, workerTask::sendRecords); verify(admin).createOrFindTopics(any()); + + verifyTopicCreation(); } @Test @@ -584,6 +599,8 @@ public void testTopicCreateFailsWithExceptionWhenCreateReturnsTopicNotCreatedOrF workerTask.toSend = Arrays.asList(record1, record2); assertThrows(ConnectException.class, workerTask::sendRecords); verify(admin).createOrFindTopics(any()); + + verifyTopicCreation(); } @Test @@ -605,6 +622,9 @@ public void testTopicCreateSucceedsWhenCreateReturnsExistingTopicFound() { List> capturedValues = sent.getAllValues(); assertEquals(2, capturedValues.size()); + + verifyTaskGetTopic(2); + verifyTopicCreation(); } @Test @@ -626,6 +646,9 @@ public void testTopicCreateSucceedsWhenCreateReturnsNewTopicFound() { List> capturedValues = sent.getAllValues(); assertEquals(2, capturedValues.size()); + + verifyTaskGetTopic(2); + verifyTopicCreation(); } private void expectSendRecord(Headers headers) { @@ -663,9 +686,12 @@ private void expectTaskGetTopic() { } private void verifyTaskGetTopic() { + verifyTaskGetTopic(1); + } + private void verifyTaskGetTopic(int times) { ArgumentCaptor connectorCapture = ArgumentCaptor.forClass(String.class); ArgumentCaptor topicCapture = ArgumentCaptor.forClass(String.class); - verify(statusBackingStore).getTopic(connectorCapture.capture(), topicCapture.capture()); + verify(statusBackingStore, times(times)).getTopic(connectorCapture.capture(), topicCapture.capture()); assertEquals("job", connectorCapture.getValue()); assertEquals(TOPIC, topicCapture.getValue());