MINOR: Clean up partition assignment logic #7249

vvcephei · 2019-08-24T00:03:12Z

These are just some "tidying up" changes I made when I was preparing to start working on KIP-441.

Committer Checklist (excluded from commit message)

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

vvcephei

@guozhangwang , what do you think about this cleanup? It mostly arose out of me trying to grok the code while getting ramped up on the KIP-441 changes.

vvcephei · 2019-08-24T00:03:45Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

    protected final Set<Integer> supportedVersions = new HashSet<>();

    private Logger log;
    private String logPrefix;
-    public enum Error {


moved to another class to shorten this file a little.

vvcephei · 2019-08-24T00:04:11Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

        }
    }

-    private static final class InternalStreamsConfig extends StreamsConfig {
-        private InternalStreamsConfig(final Map<?, ?> props) {
+    private static final class QuietStreamsConfig extends StreamsConfig {


just making this a little more readable.

We have a org.apache.kafka.streams.internals.QuietStreamsConfig to use.

Unfortunately, that's in test-utils.

I could move it to a separate internal class in this module, and then use it in test-utils, though.

Yeah, let's do that: we should try our best to avoid code duplicates.

vvcephei · 2019-08-24T00:05:01Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

@@ -412,24 +415,28 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
        }

        final boolean versionProbing;
-        if (futureMetadataVersion != UNKNOWN) {
+        if (futureMetadataVersion == UNKNOWN) {


avoiding unnecessary negation.

vvcephei · 2019-08-24T00:05:36Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

                                    } else {
-                                        numPartitionsCandidate = metadata.partitionCountForTopic(sourceTopicName);
+                                        final Integer count = metadata.partitionCountForTopic(sourceTopicName);
+                                        if (count == null) {


avoiding an unboxing warning.

The topic may not be in the metadata due to some edge conditions, we cannot throw the exception in this case.

This exception would always have happened anyway (it would have thrown an NPE on unboxing). What alternative handling would you recommend?

Hmm, in that case we can do this as-is then: I need to think through what's possible under the context of KIP-429 anyways so we can have a thorough discussion in a later PR.

vvcephei · 2019-09-03T20:28:58Z

@bbejeck , do you have time to take a look at this? It's in preparation for KIP-441 work.

guozhangwang

Made a pass over the PR. Overall I think it is a good idea to refactor StreamsPartitionAssignor to make it a smaller class while moving static functions and other classes into another class.

guozhangwang · 2019-09-03T17:30:23Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

        }
    }

-    private static final class InternalStreamsConfig extends StreamsConfig {
-        private InternalStreamsConfig(final Map<?, ?> props) {
+    private static final class QuietStreamsConfig extends StreamsConfig {


We have a org.apache.kafka.streams.internals.QuietStreamsConfig to use.

guozhangwang · 2019-09-03T17:32:15Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java


+    private String getUserEndpoint(final StreamsConfig streamsConfig) {


Just a wild thought: such functions can just be made static, and maybe we can make all of them part of a StreamsPartitionAssignorUtil class along with the AssignerError as well, if our primary goal is to make this class smaller and simpler?

Yeah, this sounds good. I was thinking about making a separate helper class for parsing out the configs.

guozhangwang · 2019-09-03T20:54:59Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

                                    } else {
-                                        numPartitionsCandidate = metadata.partitionCountForTopic(sourceTopicName);
+                                        final Integer count = metadata.partitionCountForTopic(sourceTopicName);
+                                        if (count == null) {


The topic may not be in the metadata due to some edge conditions, we cannot throw the exception in this case.

vvcephei

@guozhangwang , I've updated the PR to pull more of the logic out into helper classes.

vvcephei · 2019-09-05T18:32:29Z

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

    final String name;
    final Map<String, String> topicConfigs;

-    private int numberOfPartitions = StreamsPartitionAssignor.UNKNOWN;
+    private Optional<Integer> numberOfPartitions = Optional.empty();


This must have been a refactoring mistake at some point. "Unknown Protocol Version" is clearly the wrong constant to use to represent "Unknown Number of Partitions".

Since this isn't hot path, I've swapped it out for Optional to hopefully prevent any ambiguity.

vvcephei · 2019-09-05T18:33:04Z

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

        return numberOfPartitions;
    }

-    void setNumberOfPartitions(final int numberOfPartitions) {
+    public void setNumberOfPartitions(final int numberOfPartitions) {


This was necessary as part of breaking the StreamsPartitionAssignor into multiple classes.

vvcephei · 2019-09-05T18:35:42Z

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

-            if (existedTopicPartition.containsKey(topicName)) {
-                if (!existedTopicPartition.get(topicName).equals(numberOfPartitions)) {
+            final Optional<Integer> numberOfPartitions = entry.getValue().numberOfPartitions();
+            if (existedTopicPartition.containsKey(topicName) && numberOfPartitions.isPresent()) {


This extra condition should be effectively equivalent, since if we have an existing partition count, then it shouldn't have ever been -1 and therefore we should be fine to only run the equality check if number of partitions is also specified.

vvcephei · 2019-09-05T18:36:24Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

-    private final static int VERSION_TWO = 2;
-    private final static int VERSION_THREE = 3;
-    private final static int VERSION_FOUR = 4;
-    private final static int EARLIEST_PROBEABLE_VERSION = VERSION_THREE;


moved to separate "constants" class StreamsAssignmentProtocolVersions.

vvcephei · 2019-09-05T18:36:49Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

-    private final static int VERSION_THREE = 3;
-    private final static int VERSION_FOUR = 4;
-    private final static int EARLIEST_PROBEABLE_VERSION = VERSION_THREE;
-    protected final Set<Integer> supportedVersions = new HashSet<>();


Was only present to (unnecessarily) support test logic. I refactored the test and removed the field.

vvcephei · 2019-09-05T19:03:37Z

...ms/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfo.java

 public class AssignmentInfo {

    private static final Logger log = LoggerFactory.getLogger(AssignmentInfo.class);

-    public static final int LATEST_SUPPORTED_VERSION = 4;
-    static final int UNKNOWN = -1;


These (and the SubscriptionInfo) are logically the same as the "assignment protocol version", so I've migrated them to StreamsAssignmentProtocolVersions.

vvcephei · 2019-09-05T19:07:39Z

.../src/main/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfo.java

 public class SubscriptionInfo {

    private static final Logger log = LoggerFactory.getLogger(SubscriptionInfo.class);

-    public static final int LATEST_SUPPORTED_VERSION = 4;
-    static final int UNKNOWN = -1;


As mentioned above, these are migrated to StreamsAssignmentProtocolVersions.

vvcephei · 2019-09-05T19:11:41Z

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

@@ -119,7 +120,7 @@
    private final String userEndPoint = "localhost:8080";
    private final String applicationId = "stream-partition-assignor-test";

-    private final TaskManager taskManager = EasyMock.createNiceMock(TaskManager.class);
+    private TaskManager taskManager;


The story here is that (I think wrongly), the tests were using the task manager via the config map before, but not always initializing the mock. I guess a lot of the tests never accessed its methods before. I also guess that the mock just returns null for field accesses (specifically adminClient), but it doesn't validate the field access. Now that it's a method, it need the expectation of the call always, so we always need to initialize the mock now.

The changes I made in this class are related to this. Basically, we always initialize the taskManager mock before using it to configure the StreamsPartitionAssignor.

vvcephei · 2019-09-05T19:12:33Z

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

        EasyMock.expect(taskManager.builder()).andReturn(builder).anyTimes();
        EasyMock.expect(taskManager.prevActiveTaskIds()).andReturn(prevTasks).anyTimes();
        EasyMock.expect(taskManager.cachedTasksIds()).andReturn(cachedTasks).anyTimes();
        EasyMock.expect(taskManager.processId()).andReturn(processId).anyTimes();
-        EasyMock.replay(taskManager);


We sometimes add more expectations to the mock before using it, so I've moved the replay inline, so each method can customize their task manager.

vvcephei · 2019-09-05T19:14:07Z

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


-        final List<TaskId> activeTaskList = asList(task0, task3);


This stuff is just reorganized so that each field is defined just before it's used (so you can better tell how each field is used in the test).

vvcephei · 2019-09-05T21:01:52Z

Just for reference, the tests failed, as the copartitioning logic had too high NPath complexity.

I golfed it a bit, and I think it's significantly more readable now.

While refactoring, I noticed for the first time that it doesn't just validate the partition counts, it also sets the partition counts of the repartition topics involved in the cogroup. So, I went ahead and renamed the logic to "enforce" rather than "validate".

vvcephei · 2019-09-05T21:21:39Z

Rebased to resolve merge conflicts.

guozhangwang

LGTM! Just one nit comment otherwise LGTM.

Could we trigger a streams system test just to make sure there's no regression before merging?

guozhangwang · 2019-09-05T22:36:05Z

...main/java/org/apache/kafka/streams/processor/internals/assignment/AssignorConfiguration.java

+
+import static org.apache.kafka.common.utils.Utils.getHost;
+import static org.apache.kafka.common.utils.Utils.getPort;
+import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.LATEST_SUPPORTED_VERSION;


nit: maybe we can just merge StreamsAssignmentProtocolVersions to this class?

Thanks for the feedback. I separated them because there are lots of classes that need to reference the assignment protocol versions, but there's only one that needs to parse configurations. It would be a shame to couple knowledge of PartitionAssignor configurations into SubscriptionInfo, and AssignmentInfo, for example.

vvcephei · 2019-09-05T23:07:23Z

System tests: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/2974/
results: http://testing.confluent.io/confluent-kafka-branch-builder-system-test-results/?prefix=2019-09-06--001.1567774310--vvcephei--pre-441-cleanup--b935031/

vvcephei · 2019-09-06T14:57:29Z

Some system test failures look related. I'm digging into it.

vvcephei · 2019-09-06T21:05:36Z

Ok, finally figured out the system test failure was related to some funny business in the "FutureStreamsPartitionAssignor". I think I fixed it. Kicked off: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/2985/

vvcephei · 2019-09-07T15:25:45Z

Results: http://confluent-kafka-branch-builder-system-test-results.s3-us-west-2.amazonaws.com/2019-09-06--001.1567846807--vvcephei--pre-441-cleanup--19ee100/report.html

A couple of tests failed 2 out of 3 times. I’ll look into it when I get a chance.

vvcephei · 2019-09-08T18:48:17Z

Those tests didn't look related to me, so I ran them again, 10 times:

my branch:

http://testing.confluent.io/confluent-kafka-branch-builder-system-test-results/?prefix=2019-09-07--001.1567905224--vvcephei--pre-441-cleanup--6cc2c66/
only the StreamsBrokerBounceTest failed, 3 times

trunk:

http://testing.confluent.io/confluent-kafka-system-test-results/?prefix=2019-09-07--001.1567904855--apache--trunk--7b62248/
Only the StreamsBrokerBounceTest failed, 4 times

This looks pretty good to me, so I'd be in favor of merging, once we get one more approval.

guozhangwang · 2019-09-09T00:05:47Z

Thanks @vvcephei for the follow-up! Since this is a cleaning up I think I can just merge it myself.

vvcephei · 2019-09-09T16:02:53Z

Just for kicks, I ran the Streams test suite again x3:
http://confluent-kafka-branch-builder-system-test-results.s3-us-west-2.amazonaws.com/2019-09-08--001.1568011310--vvcephei--pre-441-cleanup--6cc2c66/report.html

This time, the EOS test failed again, twice (and the broker resilience test failed once).
It's interesting that EOS didn't fail in the x10 run, and it also hasn't been failing on trunk (which is in the context of a x1 run).

Regardless, it doesn't seem to be failing due to anything in this PR...

…ics counts (#7904) This PR fixes the regression introduced in 2.4 from 2 refactoring PRs: #7249 #7419 The bug was introduced by having a logical path leading numPartitionsCandidate to be 0, which is assigned to numPartitions and later being checked by setNumPartitions. In the subsequent check we will throw illegal argument if the numPartitions is 0. This bug is both impacting new 2.4 application and upgrades to 2.4 in certain types of topology. The example in original JIRA was imported as a new integration test to guard against such regression. We also verify that without the bug fix application will still fail by running this integration test. Reviewers: Guozhang Wang <wangguoz@gmail.com>

…ics counts (apache#7904) This PR fixes the regression introduced in 2.4 from 2 refactoring PRs: apache#7249 apache#7419 The bug was introduced by having a logical path leading numPartitionsCandidate to be 0, which is assigned to numPartitions and later being checked by setNumPartitions. In the subsequent check we will throw illegal argument if the numPartitions is 0. This bug is both impacting new 2.4 application and upgrades to 2.4 in certain types of topology. The example in original JIRA was imported as a new integration test to guard against such regression. We also verify that without the bug fix application will still fail by running this integration test. Reviewers: Guozhang Wang <wangguoz@gmail.com>

vvcephei commented Aug 24, 2019

View reviewed changes

guozhangwang reviewed Sep 3, 2019

View reviewed changes

vvcephei commented Sep 5, 2019

View reviewed changes

MINOR: Clean up StreamsPartitionAssignor

b935031

guozhangwang approved these changes Sep 5, 2019

View reviewed changes

fix system tests

6cc2c66

guozhangwang merged commit 0f177ea into apache:trunk Sep 9, 2019

vvcephei deleted the pre-441-cleanup branch September 9, 2019 16:03

abbccdda mentioned this pull request Jan 7, 2020

KAFKA-9335: Fix StreamPartitionAssignor regression in repartition topics counts #7904

Merged

3 tasks


		private String getUserEndpoint(final StreamsConfig streamsConfig) {

MINOR: Clean up partition assignment logic #7249

MINOR: Clean up partition assignment logic #7249

Conversation

vvcephei commented Aug 24, 2019

Committer Checklist (excluded from commit message)

vvcephei 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

vvcephei commented Sep 3, 2019

guozhangwang 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

vvcephei 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

vvcephei commented Sep 5, 2019

vvcephei commented Sep 5, 2019

guozhangwang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei commented Sep 5, 2019 • edited

vvcephei commented Sep 6, 2019

vvcephei commented Sep 6, 2019

vvcephei commented Sep 7, 2019

vvcephei commented Sep 8, 2019

guozhangwang commented Sep 9, 2019

vvcephei commented Sep 9, 2019

vvcephei commented Sep 5, 2019 •

edited