KAFKA-9113: Unit test for RecordCollector #5

guozhangwang · 2020-01-12T03:44:00Z

While adding the coverage I also found a couple of minor issues fixed inside RecordCollectorImpl.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

guozhangwang · 2020-01-12T18:16:49Z

@abbbccdda @ableegoldman this should be reviewed after #4

…to K9113-unit-tests-record-collector

guozhangwang · 2020-01-16T23:48:51Z

Rebased on k9113-base now.

abbccdda

Thanks, left a couple of comments

abbccdda · 2020-01-17T20:47:00Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java

@@ -301,7 +306,7 @@ private void recordSendError(final String topic, final Exception exception, fina
            if (isRecoverable(uncaughtException)) {
                // producer.send() call may throw a KafkaException which wraps a FencedException,
                // in this case we should throw its wrapped inner cause so that it can be captured and re-wrapped as TaskMigrationException
-                throw new TaskMigratedException(taskId, "Producer cannot send records anymore since it got fenced", uncaughtException);
+                throw new TaskMigratedException(taskId, "Producer cannot send records anymore since it got fenced", uncaughtException.getCause());


q: this assumes the uncaught exception is always wrapped, is that the case?

The isRecoverable will only return true if 1) itself is a KafkaException, and 2) it has a wrapped exception of ProducerFencedException or UnknownProducerIdException. So here we are certain that it has a wrapped exception and we are only interested in that exception.

abbccdda · 2020-01-17T20:51:48Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java

-    @Test
-    public void shouldNotAllowOffsetsToBeUpdatedExternally() {
-        final String topic = "topic1";
+        // should not all offsets to be modified


nit: Not all offsets should be modified.

abbccdda · 2020-01-17T20:57:05Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java

@@ -32,7 +32,7 @@
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.errors.OutOfOrderSequenceException;
+import org.apache.kafka.common.errors.AuthenticationException;


req: for L264, we didn't test the case where partition = null

abbccdda · 2020-01-17T21:11:21Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java

        );

+        final Map<TopicPartition, Long> offsets = collector.offsets();


q: so the offsets map is ever-changing with the underlying map inside collector? And this is the correct behavior as we don't need to create a deep-copy every time?

This is the old behavior. But I think it is more appropriate to not making a view but a deep-copy, will change.

abbccdda · 2020-01-17T21:13:59Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java

    }

    @Test
-    public void shouldThrowRecoverableExceptionWhenProducerForgottenInCallback() {
+    public void shouldThrowTaskMigratedExceptionOnSubsequentCallWhenProducerForgottenInCallback() {


prop: forgotten -> Unknown

abbccdda · 2020-01-17T21:26:34Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java

-    @Test(expected = StreamsException.class)
-    public void shouldThrowIfTopicIsUnknownWithDefaultExceptionHandler() {
+    @Test
+    public void shouldSwallowOnEOSAbortTxnFatalException() {


q: what does this test cover? The abortTxn call is not triggered in this case right?

Good catch! I will change the test code.

abbccdda · 2020-01-17T21:27:06Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java

@@ -742,7 +825,7 @@ public void shouldThrowIfTopicIsUnknownWithContinueExceptionHandler() {
            }
        );

-        collector.send("topic1", "3", "0", null, null, stringSerializer, stringSerializer, streamPartitioner);
+        assertThrows(StreamsException.class, () -> collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, streamPartitioner));


prop: be more specific of the thrown cause

guozhangwang · 2020-01-19T19:46:56Z

@abbccdda Addressed comments.

guozhangwang · 2020-01-19T19:49:35Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java

-        assertEquals((Long) 4L, offsets.get(new TopicPartition("topic1", 0)));
-        assertEquals((Long) 2L, offsets.get(new TopicPartition("topic1", 1)));
-        assertEquals((Long) 0L, offsets.get(new TopicPartition("topic1", 2)));
+        // with mock producer without specific partition, we would use default producer partitioner with murmur hash


@mjsax this test is added as suggested by @abbccdda , and it actually covers the case that we may want to fix later. So it is also used as a "guard" which would fail when we do not use default partitioner of producer.

mjsax · 2020-01-19T21:59:18Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java

@@ -354,7 +359,7 @@ public void close() {

    @Override
    public Map<TopicPartition, Long> offsets() {
-        return Collections.unmodifiableMap(offsets);
+        return Collections.unmodifiableMap(new HashMap<>(offsets));


Making a deep copy and making it read-only seems to be redundant to me -- why do we need both?

Consider the case where you get offsets and then read from it once, and then you tried to do sth with it later but since it is a view and its values have been changed without the caller noticing, given its call frequency I think it is safer to make the returned values final.

mjsax · 2020-01-19T22:02:49Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java

-        collector.send("topic1", "245", "0", null, null, stringSerializer, stringSerializer, streamPartitioner);
+    @Test
+    public void shouldSendWithNoPartition() {
+        final Headers headers = new RecordHeaders(new Header[]{new RecordHeader("key", "value".getBytes())});


Why do we need to pass headers in this test?

No we do not necessarily need to, do you have any concerns about this?

guozhangwang · 2020-01-22T05:39:17Z

@abbccdda @mjsax I merged this PR as is in order to unblock some further unit test fixes in #6 . Please feel free to leave further comments on it and I will address them in a follow-up commit.

This PR is collaborated by Guozhang Wang and John Roesler. It is a significant tech debt cleanup on task management and state management, and is broken down by several sub-tasks listed below: Extract embedded clients (producer and consumer) into RecordCollector from StreamTask. guozhangwang#2 guozhangwang#5 Consolidate the standby updating and active restoring logic into ChangelogReader and extract out of StreamThread. guozhangwang#3 guozhangwang#4 Introduce Task state life cycle (created, restoring, running, suspended, closing), and refactor the task operations based on the current state. guozhangwang#6 guozhangwang#7 Consolidate AssignedTasks into TaskManager and simplify the logic of changelog management and task management (since they are already moved in step 2) and 3)). guozhangwang#8 guozhangwang#9 Also simplified the StreamThread logic a bit as the embedded clients / changelog restoration logic has been moved into step 1) and 2). guozhangwang#10 Reviewers: A. Sophie Blee-Goldman <[email protected]>, Bruno Cadonna <[email protected]>, Boyang Chen <[email protected]>

fix record collector unit tests

955702c

guozhangwang added 2 commits January 13, 2020 12:55

add one more test

305417e

Merge branch 'k9113-base' of https://github.com/guozhangwang/kafka in…

ea37738

…to K9113-unit-tests-record-collector

abbccdda reviewed Jan 17, 2020

View reviewed changes

Boyang's comments

327d6a9

guozhangwang commented Jan 19, 2020

View reviewed changes

mjsax reviewed Jan 19, 2020

View reviewed changes

guozhangwang merged commit f42ffb9 into k9113-base Jan 22, 2020

guozhangwang mentioned this pull request Jan 28, 2020

KAFKA-9113: Clean up task management and state management apache/kafka#7997

Merged

3 tasks

guozhangwang deleted the K9113-unit-tests-record-collector branch April 24, 2020 23:43

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-9113: Unit test for RecordCollector #5

KAFKA-9113: Unit test for RecordCollector #5

guozhangwang commented Jan 12, 2020

guozhangwang commented Jan 12, 2020 •

edited

Loading

guozhangwang commented Jan 16, 2020

abbccdda left a comment

abbccdda Jan 17, 2020

guozhangwang Jan 19, 2020

abbccdda Jan 17, 2020

abbccdda Jan 17, 2020

guozhangwang Jan 19, 2020

abbccdda Jan 17, 2020

guozhangwang Jan 19, 2020

abbccdda Jan 17, 2020

abbccdda Jan 17, 2020

guozhangwang Jan 19, 2020

abbccdda Jan 17, 2020

guozhangwang Jan 19, 2020

guozhangwang commented Jan 19, 2020

guozhangwang Jan 19, 2020

mjsax Jan 19, 2020

guozhangwang Jan 20, 2020

mjsax Jan 19, 2020

guozhangwang Jan 21, 2020

guozhangwang commented Jan 22, 2020

		);

		final Map<TopicPartition, Long> offsets = collector.offsets();

KAFKA-9113: Unit test for RecordCollector #5

KAFKA-9113: Unit test for RecordCollector #5

Conversation

guozhangwang commented Jan 12, 2020

Committer Checklist (excluded from commit message)

guozhangwang commented Jan 12, 2020 • edited Loading

guozhangwang commented Jan 16, 2020

abbccdda left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Jan 19, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Jan 22, 2020

guozhangwang commented Jan 12, 2020 •

edited

Loading