KAFKA-7317: Use collections subscription for main consumer to reduce metadata #7969

ableegoldman · 2020-01-16T03:45:28Z

Also addresses KAFKA-8821

Note that we still have to fall back to using pattern subscription if the user has added any regex-based source nodes to the topology. Includes some minor cleanup on the side

ableegoldman · 2020-01-16T03:47:14Z

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

@@ -87,10 +85,6 @@
    // map from sink processor names to sink topic (without application-id prefix for internal topics)
    private final Map<String, String> nodeToSinkTopic = new HashMap<>();

-    // map from topics to their matched regex patterns, this is to ensure one topic is passed through on source node
-    // even if it can be matched by multiple regex patterns
-    private final Map<String, Pattern> topicToPatterns = new HashMap<>();


This is just moved since it's only used by a single class

ableegoldman · 2020-01-16T03:48:07Z

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

+                final List<String> sourceTopics = ((SourceNodeFactory) nodeFactories.get(nodeName))
+                                                      .getTopics(subscriptionUpdates);
+                nodeToSourceTopics.put(nodeName, sourceTopics);
+                sourceTopicNames.addAll(sourceTopics);


Also update sourceTopicNames to keep in sync with nodeToSourceTopics

Do we need to do this at this point? I guess so at it makes sense to have sourceTopicNames match what's in nodeToSourceTopics. I'm only asking as we never had this before and I'm curious as to why.

Well I did switch #sourceTopicPattern and the new #sourceTopicCollection to use sourceTopicNames instead of nodeToSourceTopics.values, in which case we do still need to do this.
It shouldn't matter, although I will say the hardest part of working with this code/class was figuring out which data structures did/did not contain what contents or updates. I'm inclined to leave this in so that the two similar data structures are kept in sync, but I'm fine with removing it just to keep the changes minimal/necessary

ableegoldman · 2020-01-16T03:49:39Z

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

        if (topicPattern == null) {
-            final List<String> allSourceTopics = new ArrayList<>();


This can be simplified by just using sourceTopicNames, which is identical to nodeToSourceTopics.values() but without the global topics, which we remove

ableegoldman · 2020-01-16T03:53:17Z

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

    }

+    synchronized void updateSubscribedTopics(final Set<String> topics,


The SubscriptionUpdates class and various updateXXX methods were unnecessarily complex given they all boiled down to just the 4 lines of actual actions below

ableegoldman · 2020-01-16T03:54:44Z

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

+    }
+
+    private void subscribeConsumer() {
+        if (builder.usesPatternSubscription()) {


This is the actual fix; if the user has not themselves added pattern source topics we will go back to using regular subscription (having safely disabled auto topic creation)

bbejeck

Thanks for the PR @ableegoldman. I've made a quick pass.
The build didn't run but when I ran the tests locally I get some checkstyle failures for InternalTopologyBuilder.

bbejeck · 2020-01-16T14:59:28Z

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java

@@ -1064,6 +1064,9 @@ private void checkIfUnexpectedUserSpecifiedConsumerConfig(final Map<String, Obje
        consumerProps.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, StreamsPartitionAssignor.class.getName());
        consumerProps.put(WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG, getLong(WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG));

+        // disable auto topic creation
+        consumerProps.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, "false");


qq - what does this mean for users that have enabled auto topic creation? Although it's not a best practice, this seems it could lead to unexpected behavior.

Do you mean on the server side or on the client? I agree we should consider checking for this in the user supplied configs before overriding it, but I think the reasonable default behavior is to disable it (by client config, since it's enabled server-side by default).
If users really do want it then it's on them to enable it through the main consumer config. On the other hand, prior to this it was effectively disabled permanently by the workaround of using pattern subscription. WDYT?

Yeah, that SGTM I forgot that auto topic creation is the default server-side.

bbejeck · 2020-01-16T15:01:55Z

Retest this please.

bbejeck · 2020-01-16T15:03:13Z

Strange, the PR builder didn't run on PR submission but I was able to trigger the build with "Retest..."

bbejeck · 2020-01-18T03:11:39Z

@ableegoldman failures are related - checkstyle failures.

bbejeck · 2020-01-22T15:18:51Z

retest this please

bbejeck

Thanks for this fix @ableegoldman. LGTM, I only have minor comments. We need to get a build running before merging though.

bbejeck · 2020-01-22T17:01:03Z

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

        builder.addSource(null, "source-1", null, null, null, "topic-foo");
        builder.addSource(null, "source-2", null, null, null, Pattern.compile("topic-[A-C]"));
        builder.addSource(null, "source-3", null, null, null, Pattern.compile("topic-\\d"));

-        final InternalTopologyBuilder.SubscriptionUpdates subscriptionUpdates = new InternalTopologyBuilder.SubscriptionUpdates();


Nice to get rid of the use of reflection in the tests here and below

😄Agreed!

bbejeck · 2020-01-22T17:22:09Z

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

@@ -215,6 +212,10 @@ Processor describe() {
        }
    }

+    // Map from topics to their matched regex patterns, this is to ensure one topic is passed through on source node
+    // even if it can be matched by multiple regex patterns. Only used by SourceNodeFactory
+    private final Map<String, Pattern> topicToPatterns = new HashMap<>();


I think in a previous commit this was static and I had concerns that it wouldn't work, so I'm glad to see you changed this.

bbejeck · 2020-01-22T17:35:13Z

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

+        if (hasSubscriptionUpdates()) {
+            for (final String nodeName : nodeToSourcePatterns.keySet()) {
+                //need to update nodeToSourceTopics and sourceTopicNames with topics matched from given regex
+                final List<String> sourceTopics = ((SourceNodeFactory) nodeFactories.get(nodeName))


nit: I understand the change here for compactness, but I find it a little hard to follow. This is subjective however so feel free to keep as is.

I kind of thought this way was easier to understand, but I did go back and forth on it. I'm happy to change it back

bbejeck · 2020-01-22T17:45:05Z

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

+                final List<String> sourceTopics = ((SourceNodeFactory) nodeFactories.get(nodeName))
+                                                      .getTopics(subscriptionUpdates);
+                nodeToSourceTopics.put(nodeName, sourceTopics);
+                sourceTopicNames.addAll(sourceTopics);


Do we need to do this at this point? I guess so at it makes sense to have sourceTopicNames match what's in nodeToSourceTopics. I'm only asking as we never had this before and I'm curious as to why.

bbejeck · 2020-01-22T20:22:27Z

Retest this please.

bbejeck · 2020-01-22T21:48:26Z

Retest this please.

bbejeck · 2020-01-22T21:50:57Z

Java 8 build https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/321/
Java 11 build https://builds.apache.org/job/kafka-pr-jdk11-scala2.13/4333/

~~For some reason GitHub page not always updating when build is kicked off~~

EDIT: Builds just showed up now. ???

ableegoldman · 2020-01-23T00:31:31Z

retest this please

bbejeck · 2020-01-23T01:23:13Z

Retest this please.

bbejeck · 2020-01-23T15:31:22Z

Java 8 passed
Java 11 failed with Build step 'Execute shell' marked build as failure

Retest this please.

bbejeck · 2020-01-24T01:38:49Z

Retest this please.

bbejeck · 2020-01-24T14:42:42Z

Java 8 failed with

20:39:56 # There is insufficient memory for the Java Runtime Environment to continue.
20:39:56 # Cannot create GC thread. Out of system resources.

Java 11 passed https://builds.apache.org/view/All/job/kafka-pr-jdk11-scala2.13/4355/

bbejeck · 2020-01-24T17:00:07Z

Java 11 failed with kafka.admin.DeleteConsumerGroupsTest.testDeleteCmdAllGroups
Java 8 passed

bbejeck · 2020-01-24T17:00:17Z

Retest this please.

bbejeck · 2020-01-24T19:26:52Z

Java 8 on previous PR build passed. On this run it failed on an environmental issue Caused by: java.lang.OutOfMemoryError: unable to create new native thread

Java 11 passed this run.

So I'm merging this PR.

bbejeck · 2020-01-24T19:31:13Z

Merged #7969 into trunk.

tedyu · 2020-01-24T22:59:16Z

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

    }

+    synchronized void updateSubscribedTopics(final Set<String> topics,
+                                             final String logPrefix) {
+        log.debug("{}found {} topics possibly matching subscription", logPrefix, topics);


nit: missing space between logPrefix and 'found'

topics is a Set.
What's your intention for the second parameter ?
If you want the number of topics logged, you should use topics.size().

The spacing kept bothering me too but it's actually correct (logPrefix has a space at the end) -- we should actually clean up this class at some point to use log = new LogContext(logPrefix).logger(getClass()); rather than explicitly insert the prefix everywhere.
I didn't write this log message, just moved things around, but I think the intention was to list the actual topics not just the size. I probably could've improved the wording of it though

Since topics Set can be quite large, I doubt the intention was to show the contents.
'{} topics' reads like the count of entries should be shown.

Well the topics set should only include the matching topics that the Streams app is actually subscribed to, but I agree the wording suggests it should be the size/count. Since this PR is already merged do you want to submit a quick follow-up PR?

Created #8005

tedyu · 2020-01-24T23:06:19Z

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

+        subscriptionUpdates.clear();
+        subscriptionUpdates.addAll(topics);
+
+        log.debug("{}updating builder with {} topic(s) with possible matching regex subscription(s)",


Similar comment as above two.

Conflicts and/or compiler errors due to the fact that we temporarily reverted the commit that removes Scala 2.11 support: * Exit.scala: replace SAMs with anonymous inner classes. * MiniKdc.scala: take upstream changes. # By A. Sophie Blee-Goldman (1) and others # Via Jason Gustafson * apache-github/trunk: KAFKA-9254; Overridden topic configs are reset after dynamic default change (apache#7870) MINOR: MiniKdc JVM shutdown hook fix (apache#7946) KAFKA-9152; Improve Sensor Retrieval (apache#7928) Correct exception message in DistributedHerder (apache#7995) KAFKA-7317: Use collections subscription for main consumer to reduce metadata (apache#7969) KAFKA-9181; Maintain clean separation between local and group subscriptions in consumer's SubscriptionState (apache#7941) KAFKA-7737; Use single path in producer for initializing the producerId (apache#7920) # Conflicts: # core/src/test/scala/kafka/security/minikdc/MiniKdc.scala

ableegoldman commented Jan 16, 2020

View reviewed changes

bbejeck added the streams label Jan 16, 2020

bbejeck reviewed Jan 16, 2020

View reviewed changes

bbejeck approved these changes Jan 22, 2020

View reviewed changes

ableegoldman added 6 commits January 23, 2020 14:13

add consumer config and change subscription type

c75d302

side cleanup

4bbab82

replace null check in other classes with added method

ad55ffc

fix up tests, add unit test

a956735

fix checkstyle

3b52129

github comment

9ba3ee4

ableegoldman force-pushed the 7317-use-regular-subscription branch from 05dc812 to 9ba3ee4 Compare January 23, 2020 22:13

bbejeck merged commit 57b2f68 into apache:trunk Jan 24, 2020

tedyu reviewed Jan 24, 2020

View reviewed changes

tedyu mentioned this pull request Jan 25, 2020

MINOR: Fix topology builder debug log message #8005

Merged

3 tasks

ableegoldman deleted the 7317-use-regular-subscription branch June 26, 2020 22:38

		if (topicPattern == null) {
		final List<String> allSourceTopics = new ArrayList<>();

		}

		synchronized void updateSubscribedTopics(final Set<String> topics,

KAFKA-7317: Use collections subscription for main consumer to reduce metadata #7969

KAFKA-7317: Use collections subscription for main consumer to reduce metadata #7969

Conversation

ableegoldman commented Jan 16, 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

Choose a reason for hiding this comment

ableegoldman Jan 16, 2020 • edited Loading

Choose a reason for hiding this comment

bbejeck 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

bbejeck commented Jan 16, 2020

bbejeck commented Jan 16, 2020

bbejeck commented Jan 18, 2020

bbejeck commented Jan 22, 2020

bbejeck left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ableegoldman Jan 23, 2020 • edited Loading

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

bbejeck commented Jan 22, 2020

bbejeck commented Jan 22, 2020

bbejeck commented Jan 22, 2020 • edited Loading

ableegoldman commented Jan 23, 2020

bbejeck commented Jan 23, 2020

bbejeck commented Jan 23, 2020

bbejeck commented Jan 24, 2020

bbejeck commented Jan 24, 2020

bbejeck commented Jan 24, 2020

bbejeck commented Jan 24, 2020

bbejeck commented Jan 24, 2020

bbejeck commented Jan 24, 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ableegoldman Jan 16, 2020 •

edited

Loading

ableegoldman Jan 23, 2020 •

edited

Loading

bbejeck commented Jan 22, 2020 •

edited

Loading