KAFKA-10357: Extract setup of repartition topics from Streams partition assignor by cadonna · Pull Request #9848 · apache/kafka

cadonna · 2021-01-08T18:58:10Z

To implement the explicit user initialization of Kafka Streams as
described in KIP-698, we first need to extract the code for the
setup of the repartition topics from the Streams partition assignor
so that it can also be called outside of a rebalance.

Committer Checklist (excluded from commit message)

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

…tition assignor To implement the explicit user initialization of Kafka Streams as described in KIP-698, we first need to extract the code for the setup of the repartition topics from the Streams partition assignor so that it can also be called outside of a rebalance.

cadonna · 2021-01-08T18:59:22Z

Call for review: @ableegoldman @guozhangwang @lct45 @wcarlson5

lct45 · 2021-01-12T14:06:52Z

+        return Collections.unmodifiableMap(topicPartitionInfos);
+    }
+
+    private Map<String, InternalTopicConfig> computeRepartitionTopicConfig(final Map<Integer, TopicsInfo> topicGroups,


Is there a reason you renamed this from computeRepartitionTopicMetadata to computeRepartitionTopicConfig?

I thought since the returned value is a bunch of configs computeRepartitionTopicConfig() would be more precise.

lct45

Overall, LGTM. Looks like most of the logic has stayed consistent with what was there. Thanks for the PR @cadonna !

lct45 · 2021-01-12T14:16:56Z

+
+                        if (numPartitions == null) {
+                            partitionCountNeeded = true;
+                            log.trace("Unable to determine number of partitions for {}, another iteration is needed",


Maybe I'm missing something, but it looks like we want to go through the do loop again if we have null partitions, but in this case wouldn't we hit the if statement below and throw an exception, since at this point partitionCountNeeded == True and progressMadeThisIteration == False ? Or does that just log an exception and continue through the do loop?

@lct45 sort of; the point of progressMadeThisIteration is to break out of the loop entirely if we detect that we're infinitely looping, as was the case in a recent-ish bug. We want to continue looping only as long as we're actually gaining new information/filling in the partition count of some topic group (subtopology).

So yes, if progressMadeThisIteration = false after the full loop, but we still don't have all the partition counts (partitionCountNeeded = true), then oh well. Something went wrong 🙂 We should throw

cadonna · 2021-01-11T15:28:32Z

    private final Node[] inSyncReplicas;
    private final Node[] offlineReplicas;

-    // Used only by tests


The usual lying comment. This constructor is not only used in tests.

ableegoldman

Nice work. A few of the variable names were kind of throwing me off, but you can apply my suggestions (or not :P) in a followup PR if you'd prefer. This one pretty much LGTM, looking forward to the next PR

ableegoldman · 2021-01-12T20:18:51Z

+
+    private void checkIfExternalSourceTopicsExist(final TopicsInfo topicsInfo,
+                                                  final Cluster clusterMetadata) {
+        final Set<String> externalSourceTopics = new HashSet<>(topicsInfo.sourceTopics);


prop: rename to missingExternalSourceTopics or similar

ableegoldman · 2021-01-13T03:26:34Z

+
+                        if (numPartitions == null) {
+                            partitionCountNeeded = true;
+                            log.trace("Unable to determine number of partitions for {}, another iteration is needed",


@lct45 sort of; the point of progressMadeThisIteration is to break out of the loop entirely if we detect that we're infinitely looping, as was the case in a recent-ish bug. We want to continue looping only as long as we're actually gaining new information/filling in the partition count of some topic group (subtopology).

So yes, if progressMadeThisIteration = false after the full loop, but we still don't have all the partition counts (partitionCountNeeded = true), then oh well. Something went wrong 🙂 We should throw

ableegoldman · 2021-01-13T03:39:39Z

+                                          final String repartitionSourceTopic) {
+        Integer partitionCount = null;
+        // try set the number of partitions for this repartition topic if it is not set yet
+        for (final TopicsInfo upstreamTopicsInfo : topicGroups.values()) {


It's a bit confusing to call this upstreamTopicsInfo since that will only be true of one (or some) of these topic groups. Most of them are not upstream of the current

Yeah, I was also not sure. I renamed some of the variables. Let me know if it is better now.

ableegoldman · 2021-01-13T03:46:10Z

+    final Cluster clusterMetadata = niceMock(Cluster.class);
+
+    @Before
+    public void setUp() {


Did you mean to leave in an empty method?

No, I will remove it.

ableegoldman · 2021-01-13T03:57:59Z

+            metadata,
+            logPrefix
+        );
+        repartitionTopics.setup();


Why not just do the setup inside the constructor?

That is a valid question. When we introduce the explicit initialization and the manual and automatic configs, we need to distinguish between setting up the internal topics and verifying the internal topics. Hence, there will be two methods on the RepartitionTopic class. We will create the object and then according to the configs we will either just verify or verify and setup the repartition topics. We could accomplish the same with a flag in the constructor but I thought the code might be easier to understand with two methods.

Makes sense, thanks

ableegoldman · 2021-01-15T21:07:56Z

@cadonna looks like checkstyleTest is failing

cadonna · 2021-01-18T12:07:11Z

@ableegoldman Thank you for the info!
This happens when you try to take shortcuts Friday evening. 😁

guozhangwang

Made a pass on the PR. Overall LGTM.

I think once they current comments are addressed it can be merged cc @ableegoldman

guozhangwang · 2021-01-19T05:07:23Z

            final Map<Integer, TopicsInfo> topicGroups = taskManager.builder().topicGroups();

-            final Map<TopicPartition, PartitionInfo> allRepartitionTopicPartitions = prepareRepartitionTopics(topicGroups, metadata);
+            final Map<TopicPartition, PartitionInfo> allRepartitionTopicPartitions = prepareRepartitionTopics(metadata);


nit: line 365 above can be moved down now since it is only needed before line 379.

guozhangwang · 2021-01-19T05:12:04Z

+
+        // make sure the repartition source topics exist with the right number of partitions,
+        // create these topics if necessary
+        internalTopicManager.makeReady(repartitionTopicMetadata);


Not related to this PR: when reading the makeReady code again, I feel that the way we get the retries config by first constructing a dummy admin client can be simplified. Also, it is debatable that if we only do verify, then do we really need to retry or not since the admin client itself already retries internally.

I think we should keep this in mind. For this ticket, we need to touch makeReady() anyways, because we need to separate verification from creation.

guozhangwang · 2021-01-19T05:12:31Z

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals;


As always, thanks for the great test coverage!

cadonna · 2021-01-22T10:24:59Z

@ableegoldman @guozhangwang, could someone of you merge this PR since it seems there are no further open question.

ableegoldman · 2021-01-22T18:01:56Z

One unrelated test failure in SocketServerTest.testConnectionRatePerIp

ableegoldman · 2021-01-22T18:03:45Z

Merged to trunk

…e-allocations-lz4 * apache-github/trunk: (562 commits) MINOR: remove unused code from MessageTest (apache#9961) MINOR: Fix visibility of Log.{unflushedMessages, addSegment} methods (apache#9966) KAFKA-12229: Restore original class loader in integration tests using EmbeddedConnectCluster during shutdown (apache#9942) KAFKA-12190: Fix setting of file permissions on non-POSIX filesystems (apache#9947) MINOR: Remove `toStruct` and `fromStruct` methods from generated protocol classes (apache#9960) MINOR: Fix typo in Utils#toPositive (apache#9943) MINOR: MessageUtil: remove some deadcode (apache#9931) MINOR: Update zstd-jni to 1.4.8-2 (apache#9957) MINOR: Revert assertion in MockProducerTest (apache#9956) MINOR: Optimize assertions in unit tests (apache#9955) MINOR: Tag `RaftEventSimulationTest` as `integration` and tweak it (apache#9925) MINOR: Update to Gradle 6.8.1 (apache#9953) MINOR: A few small group coordinator cleanups (apache#9952) MINOR: Upgrade ducktape to version 0.8.1 (apache#9933) MINOR: fix record time in test shouldWipeOutStandbyStateDirectoryIfCheckpointIsMissing (apache#9948) MINOR: Restore interrupt status when closing (apache#9863) KAFKA-10357: Extract setup of repartition topics from Streams partition assignor (apache#9848) KAFKA-12212; Bump Metadata API version to remove `ClusterAuthorizedOperations` fields (KIP-700) (apache#9945) MINOR: log 2min processing summary of StreamThread loop (apache#9941) MINOR: Drop enable.metadata.quorum config (apache#9934) ...

Remove interface for cluster metadata

c18fbad

lct45 reviewed Jan 12, 2021

View reviewed changes

cadonna added 2 commits January 12, 2021 21:17

Restrict scope of PR

f879bf8

Remove some unnecessary refactorings

0f06eca

cadonna commented Jan 12, 2021

View reviewed changes

ableegoldman approved these changes Jan 13, 2021

View reviewed changes

ableegoldman reviewed Jan 13, 2021

View reviewed changes

cadonna added 2 commits January 15, 2021 17:59

Rename some variables

7b1f28d

Remove empty test setup method

0de63c6

Resolve checkstyle issue

195e585

guozhangwang reviewed Jan 19, 2021

View reviewed changes

Move topic groups retrieval a couple of code lines down

f29d690

ableegoldman merged commit 019cd4a into apache:trunk Jan 22, 2021

Conversation

cadonna commented Jan 8, 2021

Committer Checklist (excluded from commit message)

Uh oh!

cadonna commented Jan 8, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lct45 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ableegoldman left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ableegoldman commented Jan 15, 2021

Uh oh!

cadonna commented Jan 18, 2021

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cadonna commented Jan 22, 2021

Uh oh!

ableegoldman commented Jan 22, 2021

Uh oh!

ableegoldman commented Jan 22, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants