KAFKA-6150: KIP-204 part III; Change repartition topic segment size and ms #4315

guozhangwang · 2017-12-12T01:29:29Z

Create default internal topic configs in StreamsConfig, especially for repartition topics change the segment size and time to smaller value.
Consolidate the default internal topic settings to InternalTopicManager and simplify InternalTopicConfig correspondingly.
Add an integration test for purging data.
MINOR: change TopologyBuilderException to IllegalStateException in StreamPartitionAssignor (part of https://issues.apache.org/jira/browse/KAFKA-5660).

Here are a few public facing APIs that get added:

AbstractConfig#originalsWithPrefix(String prefix, boolean strip): this for simplify the logic of passing admin and topic prefixed configs to consumer properties.
KafkaStreams constructor with Time object for convienent mocking in tests.

Will update KIP-204 accordingly if people re-votes these changes.

Committer Checklist (excluded from commit message)

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

- fixed parameter passing for `retries` - added couple of tests

into K6150-segment-size

…into K6150-segment-size

guozhangwang · 2017-12-12T02:35:46Z

@dguy @mjsax @bbejeck for reviews.

dguy

Thanks @guozhangwang, made a quick pass and left a few comments/questions

dguy · 2017-12-12T16:38:59Z

clients/src/main/java/org/apache/kafka/clients/admin/NewTopic.java

+    /**
+     * The configuration for the new topic or null if no configs ever specified.
+     */
+    public Map<String, String> configs() {


public API change? is this in the KIP?

Yes I'm planning to update the KIP and calls for a re-vote including this change.

dguy · 2017-12-12T16:46:04Z

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

+
+                topicConfig.putAll(internalTopicConfig.toProperties(windowChangeLogAdditionalRetention));
+
+                System.out.println("\nCREATE topic " + internalTopicConfig.name() + " with partitions " +  internalTopicConfig.numberOfPartitions() + ": config " + topicConfig + "\n");


trace log maybe?

dguy · 2017-12-12T16:48:23Z

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

+                // internal topic config overridden rule: library overrides < global config overrides < per-topic config overrides
+                final Map<String, String> topicConfig = new HashMap<>();
+
+                switch (internalTopicConfig.type()) {


This and other code, i.e, if(topicType == ..) makes me think we should maybe have an interface and three implementations of this. Then we can remove the if and switch around the type and just ask the impl to do whatever it needs to

Not sure I understand this statement, could you elaborate a bit more? Currently this switch pattern is only called once so I'm not sure if it is worthwhile to refactor it as an interface, or maybe I completely got it wrong since you mention this and other code (while I cannot think of other code?).

I think what Damian meant is, that we need to pass in the internal topic type in the constructor -- instead of that, we can just instantiate different types directly.

I see your point now. The reason I wanted to pass in the topicType at constructor is that, the setRetentionMs() can be guarded earlier like we did today: if we do not know the topicType we will blindly accept the retention ms.

dguy · 2017-12-12T16:50:09Z

...src/test/java/org/apache/kafka/streams/integration/PurgeRepartitionTopicIntegrationTest.java

+               .groupBy(MockKeyValueMapper.SelectKeyKeyValueMapper())
+               .count();
+
+        System.out.println(builder.build().describe());


i guess this isn't needed

…into K6150-segment-size

mjsax

Did a pass. Couple of minor comments.

mjsax · 2017-12-12T20:53:47Z

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

+     * @param time           {@code Time} implementation; cannot be null
+     * @throws StreamsException if any fatal error occurs
+     */
+    public KafkaStreams(final Topology topology,


Public API change! Can't do this. Can this be package private? Why do we need this?

Yes, see my description on the PR.

You mentioned in the PR description that Time was added for the convenience of unit testing, should we add something similar to javadoc?

We usually do not need to mention that, since people may have their own Time implementation other than SYSTEM in their applications that is not only for testing.

mjsax · 2017-12-12T20:54:33Z

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

    private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
                         final StreamsConfig config,
                         final KafkaClientSupplier clientSupplier) throws StreamsException {
+        this(internalTopologyBuilder, config, clientSupplier, Time.SYSTEM);


can't we remove this private constructor and add Time.SYSTEM to the internal this() calls of the others?

We can, though it means we need to add this Time.SYSTEM on all the callers, whereas I was trying to consolidate these in a single call. I guess this this pure subjective. Let me know if you have a strong opinion.

No strong opinion.

mjsax · 2017-12-12T20:56:37Z

clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java

+     * @param strip strip the prefix before adding to the output if set true
+     * @return a Map containing the settings with the prefix
+     */
+    public Map<String, Object> originalsWithPrefix(String prefix, boolean strip) {


Seems like a public API change.

Yes, see my description on the PR.

Ah. Reading. A skill I still need to master... :)

mjsax · 2017-12-12T21:02:32Z

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

+        // add admin and topic configs required for creating topics
+        consumerProps.put(adminClientPrefix(RETRIES_CONFIG), getInt(RETRIES_CONFIG));
+        consumerProps.putAll(originalsWithPrefix(TOPIC_PREFIX, false));
+        consumerProps.putAll(originalsWithPrefix(ADMIN_CLIENT_PREFIX, false));


Do we need to pass in all admin client configs? Actually, we only need retries.

Also, I am not sure if we would set the correct default here. If nothing is specified, we would use StreamsConfig retry as default, but we actually should set AdminClientConfig default -- atm, both might be the same so it doesn't matter too much. Just is doesn't seems to be "correct" (in a very strong sense).

That's a good point: I will revert the changes.

mjsax · 2017-12-12T21:05:07Z

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

-                builder.append(cleanupPolicy.name()).append(",");
-            }
-            builder.deleteCharAt(builder.length() - 1);
+    public Map<String, String> toProperties(final long additionalRetentionMs) {


Nit: Return type and method name to does match. maybe rename to addToTopicConfig ?

Also update JavaDoc return tag

I think the name / return tag still makes sense: it returns a map of properties from its topic config, retention ms and the passed-in additional retention ms.

Sound like it uses Properties class (while it doesn't) -- but maybe I am overthinking this... Should be fine.

mjsax · 2017-12-12T21:42:01Z

...src/test/java/org/apache/kafka/streams/integration/PurgeRepartitionTopicIntegrationTest.java

+            try {
+                Collection<DescribeLogDirsResponse.LogDirInfo> logDirInfo = adminClient.describeLogDirs(Collections.singleton(0)).values().get(0).get().values();
+
+                if (logDirInfo.isEmpty()) {


WE don't need this check. If logDirInfo.isEmpty() == true the for-loop will never be entered and we hit the return false after the loop.

mjsax · 2017-12-12T21:44:35Z

...src/test/java/org/apache/kafka/streams/integration/PurgeRepartitionTopicIntegrationTest.java

+                "Repartition topic " + REPARTITION_TOPIC + " not created with the expected configs after 60000 ms.");
+
+        TestUtils.waitForCondition(
+                new RepartitionTopicPurged(new TopicSizeVerifier() {


If I understand this correctly, we actually test if the topic is not purged yet? Thus, the name RepartitionTopicPurged is a little miss leading/confusing.

Ack. I'm renaming to RepartitionTopicVerified, which is used for both verify filled and verify purged.

mjsax · 2017-12-12T21:47:42Z

...src/test/java/org/apache/kafka/streams/integration/PurgeRepartitionTopicIntegrationTest.java

+                    }
+                }
+                return false;
+            } catch (Exception e) {


nit: add final

mjsax · 2017-12-12T21:48:19Z

...src/test/java/org/apache/kafka/streams/integration/PurgeRepartitionTopicIntegrationTest.java

+                        }
+                    }
+                }
+                return false;


can't we have single return false at the very end?

mjsax · 2017-12-12T21:50:45Z

...src/test/java/org/apache/kafka/streams/integration/PurgeRepartitionTopicIntegrationTest.java

+                new RepartitionTopicPurged(new TopicSizeVerifier() {
+                    @Override
+                    public boolean verify(long currentSize) {
+                        return currentSize <= purgeSegmentBytes;


Must it not be < ? If the latest segment is full, won't we roll it and thus it would get purged?

We will roll if the segment size > segment bytes, not >=.

mjsax · 2017-12-13T00:13:04Z

...src/test/java/org/apache/kafka/streams/integration/PurgeRepartitionTopicIntegrationTest.java

+
+        @Override
+        public final boolean conditionMet() {
+            time.sleep(purgeIntervalMs);


But we advance the time before we verify that the topic was written. I think, this is ok for the "purging" test but not for the first test for which we expect to see data in the repartitioning topic. As the StreamsThreads run in the back ground, in the first check directly after we advanced time, we could have a context switch and data might get purged and thus, we can never hit the condition that "repartitionsTopicSize > 0"

Or can this never happen?

This should never happen, since starting offset based purging should never be "clean". I.e. even if you set the purge offset to be LEO it should not delete the topic partition segments to empty.

…into K6150-segment-size

guozhangwang · 2017-12-13T23:50:13Z

Jenkins failures seem not relevant. @dguy @mjsax have addressed the comments.

mjsax · 2017-12-14T18:24:36Z

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

@@ -112,6 +112,8 @@
     * These should be valid properties from {@link org.apache.kafka.common.config.TopicConfig TopicConfig}.
     * It is recommended to use {@link #topicPrefix(String)}.
     */
+    // TODO: currently we cannot get the full topic configurations and hence cannot allow topic configs without the prefix,
+    //       this can be lifted once kafka.log.LogConfig is completely deprecated by org.apache.kafka.common.config.TopicConfig


Don't understand this comment. Can you elaborate?

For admin, producer and consumer we have getXXXConfigs which relies on getClientPropsWithPrefix(XXX_PREFIX, XXXConfig.configNames()); However for topic configs we do not have such a class (note TopicConfig is not sufficient, and the source of truth is still in broker-side KafkaConfig), so we cannot have a getTopicConfigs that can detect all the topic config names. Hence we ALWAYS have to prefix the config name to indicate it is a topic config name, unlike others which we do not have to.

Thx. Got it.

mjsax · 2017-12-14T18:26:23Z

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

        final AdminClientConfig config = new AdminClientConfig(getClientPropsWithPrefix(ADMIN_CLIENT_PREFIX, AdminClientConfig.configNames()));
        consumerProps.put(adminClientPrefix(AdminClientConfig.RETRIES_CONFIG), config.getInt(AdminClientConfig.RETRIES_CONFIG));

+        // add admin and topic configs required for creating topics, also add batch size of producer for verification
+        final Map<String, Object> producerProps = getClientPropsWithPrefix(PRODUCER_PREFIX, ProducerConfig.configNames());
+        consumerProps.put(producerPrefix(ProducerConfig.BATCH_SIZE_CONFIG), producerProps.containsKey(ProducerConfig.BATCH_SIZE_CONFIG) ? (Integer) producerProps.get(ProducerConfig.BATCH_SIZE_CONFIG) : 16384);


Where does 16384 come from? Sounds like the producer default config value. Shouldn't we get this out of the ProducerConfig similar to retires from AdminClientConfig ?

Yes it is indeed from ProducerConfig, unfortunately this value is not exposed from ProducerConfig object and hence cannot be leveraged.

Yes. We have the same issue with AdminClientConfig and retries -- thus, we instantiate a AdminClientConfig got get the default out of it. Can we do the same thing here and instantiate a ProducerConfig object?

I now it's not very nice code, but still better than hardcoding the value.

mjsax · 2017-12-14T18:29:17Z

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

-                    topicConfig.put(key, topicProperties.getProperty(key));
+                final Map<String, String> topicConfig = internalTopicConfig.getProperties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
+
+                if (topicConfig.containsKey(TopicConfig.SEGMENT_INDEX_BYTES_CONFIG) &&


Can't we put this check into StreamsConfig ?

I had tried that. The problem is that the default topic configs value are unknown at the beginning. But maybe I can rework something out of it.

Bare with me for another minor change.

@mjsax Okay now I remembered why I cannot do that, because ProducerConfig constructor is not public in its package as AdminClientConfig constructor. That's why I cannot do sth. like

new ProducerConfig(getProducerConfigs("dummy")).getInt(ProducerConfig.BATCH_SIZE_CONFIG);

to read it out.

Sounds like an inconsistency... Would it be a problem making those constructors (ie, also for consumer) public?

dguy · 2017-12-15T13:00:51Z

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

+        // internal topic config overridden rule: library overrides < global config overrides < per-topic config overrides
+        final Map<String, String> topicConfig = new HashMap<>();
+
+        switch (topicType) {


I think i said this before, but i think it would be better/nicer to have different implementations of InternalTopicConfig for the different types of internal topics and then we don't need to have switch and if statements. Each impl will know what ti needs to do. WDYT?

I thought about it but felt it may be an overkill.. because now we have all the logics depending on type consolidated in one class already.

I personally support Damian idea here. Don't think it's overkill but actually more elegant OO code. If you feel strong to keep as it, also fine with me though.

@dguy @mjsax Just adding a few more notes here: I think the major concern is our type-dependent logic is scattered across multiple packages and classes, so we'd like to consolidate it. One way is to still pass in the type information but make all related logic within this single class (what I did), another way is to make different instantiations of the class so we do not need to have this type field at all. Personally I think they achieved similar goal while mine is less code: the downside is, as mentioned in #4315 (comment) the setRetentionMs should only be accepted with windowed store changelog type, but we still need to provide this API for either approach.

I personally support Damian idea here. Don't think it's overkill but actually more elegant OO code. If you feel strong to keep as it, also fine with me though.

I did not see this comment while writing mine. If my understanding about the root concern is right but we feel it is better to get rid of this type information, I can change it (i.e. I do not feel strong about it :P). LMK @mjsax @dguy

I don't think it is overkill either, it is just good OO design. Anyway, i'm not going to block this PR because of it.

Cool, I have refactored InternalTopicConfig in the latest commit as you suggested and will go ahead to merge it as is; if you can take a quick look that would be greatly appreciated.

…into K6150-segment-size

guozhangwang · 2017-12-18T19:36:20Z

@mjsax @dguy could you take another look? Also please let me know about #4315 (comment)

mjsax

LGTM.

mjsax · 2017-12-18T22:48:37Z

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

-        consumerProps.put(producerPrefix(ProducerConfig.BATCH_SIZE_CONFIG), producerProps.containsKey(ProducerConfig.BATCH_SIZE_CONFIG) ? (Integer) producerProps.get(ProducerConfig.BATCH_SIZE_CONFIG) : 16384);
-        consumerProps.putAll(originalsWithPrefix(TOPIC_PREFIX, false));
+        // verify that producer batch config is no larger than segment size, then add topic configs required for creating topics
+        Map<String, Object> topicProps = originalsWithPrefix(TOPIC_PREFIX, false);


Nit: add final

dguy

LGTM

…into K6150-segment-size

guozhangwang · 2017-12-20T00:03:00Z

Jenkins failures are irrelevant, it is a known issue: https://issues.apache.org/jira/browse/KAFKA-6377

guozhangwang · 2017-12-20T00:06:04Z

Merged to trunk.

mjsax · 2017-12-20T22:29:17Z

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

-    private final Map<String, String> logConfig;
-    private final Set<CleanupPolicy> cleanupPolicies;
+    // we need to distinguish windowed and un-windowed store changelog since their cleanup policy may be different
+    public enum InternalTopicType { REPARTITION, WINDOWED_STORE_CHANGELOG, UNWINDOWED_STORE_CHANGELOG }


I guess this can be removed now?

Ack, will do in the docs PR.

tedyu · 2017-12-21T21:53:48Z

clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java

-            if (entry.getKey().startsWith(prefix) && entry.getKey().length() > prefix.length())
-                result.put(entry.getKey().substring(prefix.length()), entry.getValue());
+            if (entry.getKey().startsWith(prefix) && entry.getKey().length() > prefix.length()) {
+                if (strip)


It seems this check can be lifted above the check at line 192.
Since we don't need to check prefix if strip == false.

Actually we still need to check prefix even if strip == false, the only difference is that if the check of the prefix succeeds, whether or not to maintain the prefix in the returned map.

mjsax and others added 11 commits December 6, 2017 12:41

KAFKA-4857: Replace StreamsKafkaClient with AdminClient in Kafka Streams

7f39481

Github comments

49e832c

- fixed parameter passing for `retries` - added couple of tests

rebased

ad7ccde

fixed system test error message

bfd3937

Merge branch 'kafka-4857-admit-client' of https://github.com/mjsax/kafka

81b17ba

into K6150-segment-size

add internal topic types

3c286b9

use default props for internal topics

6c223b8

Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka …

177d11a

…into K6150-segment-size

add integration test for purging repartition topic

ef57e80

Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka …

7bf6553

…into K6150-segment-size

augment unit tests

b2f3a21

dguy reviewed Dec 12, 2017

View reviewed changes

Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka …

8b0a567

…into K6150-segment-size

mjsax reviewed Dec 12, 2017

View reviewed changes

github comments

574f069

mjsax reviewed Dec 13, 2017

View reviewed changes

guozhangwang added 3 commits December 12, 2017 18:29

revert admin configs

d9face0

Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka …

7fd7c5e

…into K6150-segment-size

github comments round two

2d46704

mjsax reviewed Dec 14, 2017

View reviewed changes

dguy reviewed Dec 15, 2017

View reviewed changes

guozhangwang added 2 commits December 15, 2017 11:12

Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka …

9c583df

…into K6150-segment-size

Move batch size checking to StreamsConfig

96d4e82

mjsax mentioned this pull request Dec 18, 2017

KAFKA-6382: Make ProducerConfig and ConsumerConfig constructors public #4341

Merged

mjsax approved these changes Dec 18, 2017

View reviewed changes

dguy approved these changes Dec 19, 2017

View reviewed changes

Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka …

2960486

…into K6150-segment-size

Refactor InternalTopicConfig

2f5db84

asfgit closed this in 82c6d42 Dec 20, 2017

mjsax reviewed Dec 20, 2017

View reviewed changes

tedyu reviewed Dec 21, 2017

View reviewed changes

guozhangwang deleted the K6150-segment-size branch February 14, 2018 19:50


		topicConfig.putAll(internalTopicConfig.toProperties(windowChangeLogAdditionalRetention));

		System.out.println("\nCREATE topic " + internalTopicConfig.name() + " with partitions " + internalTopicConfig.numberOfPartitions() + ": config " + topicConfig + "\n");

KAFKA-6150: KIP-204 part III; Change repartition topic segment size and ms #4315

KAFKA-6150: KIP-204 part III; Change repartition topic segment size and ms #4315

Conversation

guozhangwang commented Dec 12, 2017 • edited Loading

Committer Checklist (excluded from commit message)

guozhangwang commented Dec 12, 2017

dguy 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

mjsax 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

mjsax Dec 12, 2017 • 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

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 Dec 13, 2017

Choose a reason for hiding this comment

guozhangwang Dec 15, 2017 • 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

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 Dec 18, 2017

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dguy left a comment

Choose a reason for hiding this comment

guozhangwang commented Dec 20, 2017

guozhangwang commented Dec 20, 2017

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 Dec 12, 2017 •

edited

Loading

mjsax Dec 12, 2017 •

edited

Loading

guozhangwang Dec 15, 2017 •

edited

Loading