You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "C0urante (via GitHub)" <gi...@apache.org> on 2023/02/06 20:03:30 UTC
[GitHub] [kafka] C0urante commented on a diff in pull request #13191: KAFKA-14060: Replace EasyMock and PowerMock with Mockito in AbstractWorkerSourceTaskTest
C0urante commented on code in PR #13191:
URL: https://github.com/apache/kafka/pull/13191#discussion_r1097852548
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java:
##########
@@ -129,7 +131,7 @@ public class AbstractWorkerSourceTaskTest {
@Mock private ConnectorOffsetBackingStore offsetStore;
@Mock private StatusBackingStore statusBackingStore;
@Mock private WorkerSourceTaskContext sourceTaskContext;
- @MockStrict private TaskStatus.Listener statusListener;
Review Comment:
In order to retain the same guarantees we have currently w/r/t interactions with this class, can we add a call to `verifyNoMoreInteractions(statusListener);` in the `tearDown` method?
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTaskTest.java:
##########
@@ -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<NewTopic> 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<SourceRecord>) invocation -> invocation.getArgument(0));
+ 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);
+ });
+ 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<ProducerRecord<byte[], byte[]>> expectSendRecord(
- String topic,
- boolean anyTimes,
- Headers headers
- ) {
- if (headers != null)
- expectConvertHeadersAndKeyValue(topic, anyTimes, headers);
-
- expectApplyTransformationChain(anyTimes);
-
- Capture<ProducerRecord<byte[], byte[]>> sent = EasyMock.newCapture();
-
- IExpectationSetters<Future<RecordMetadata>> expect = EasyMock.expect(
- producer.send(EasyMock.capture(sent), EasyMock.capture(producerCallbacks)));
-
- IAnswer<Future<RecordMetadata>> 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<ProducerRecord<byte[], byte[]>> expectSendRecordAnyTimes() {
- return expectSendRecord(TOPIC, true, emptyHeaders());
- }
-
- private Capture<ProducerRecord<byte[], byte[]>> expectSendRecord() {
- return expectSendRecord(TOPIC, false, emptyHeaders());
- }
-
- 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 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));
- }
Review Comment:
Why remove this method? It seems like now there's a lot more duplication in each test case that used to call it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org