KAFKA-8305: support default partitions & replication factor in AdminClient#createTopic #6728

agavra · 2019-05-14T16:53:20Z

See: KIP-464 for more information.

Description

This change makes the two required changes to support creating topics using the cluster defaults for replication and partitions:

Adds a NewTopic(String) constructor to the NewTopic API
Changes the AdminManager to accept -1 as valid options for replication factor and partitions. If this is the case, it will resolve it using the default configuration.
This also adds a dependency on scalaJava8Compat library to make it simpler to convert Scala Option to Optional.

Testing

Updated unit tests with the new conditions

Committer Checklist (excluded from commit message)

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

ryannedolan

Looking forward to this!

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

ijuma

Thanks for the PR, I had a quick look and left some comments.

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

ijuma · 2019-05-15T05:18:46Z

core/src/main/scala/kafka/server/AdminManager.scala

            val replicationFactor: java.lang.Short =
-              if (topic.replicationFactor == NO_REPLICATION_FACTOR) null else topic.replicationFactor
+              if (topic.replicationFactor == NO_REPLICATION_FACTOR) null else resolvedReplicationFactor


The documentation for CreateTopicPolicy says:

@param numPartitions the number of partitions to create or null if replicasAssignments is set.
@param replicationFactor the replication factor for the topic or null if replicaAssignments is set.

It seems like we may end up with a non null numPartitions and replicationFactor even if replicaAssignment is set?

What's the scenario that causes this? If replicaAssignments is set, then we expect the partitions/replication factor to equal NO_PARTITIONS and NO_REPLICATION_FACTOR respectively, otherwise it fails here:

if ((topic.numPartitions != NO_NUM_PARTITIONS || topic.replicationFactor != NO_REPLICATION_FACTOR) && !topic.assignments().isEmpty) { throw new InvalidRequestException("Both numPartitions or replicationFactor and replicasAssignments were set. " + "Both cannot be used at the same time.") }

You're right. But then why do we need this change at all? It seems like we never end up with a different result than the previous code.

The difference is that if the value sent was -1 (i.e. NO_REPLICATION_FACTOR) and there are no assignments, it will instead be set the resolvedReplicationFactor, which is the cluster default (i.e. default.replication.factor)

(you can see tests for the changed behavior)

Hmm.. I think Ismael's point is that if topic.replicationFactor != NO_REPLICATION_FACTOR, then resolvedReplicationFactor == topic.replicationFactor, and this is equivalent to what we already had. It seems like this is what we're trying to do:

val replicationFactor: java.lang.Short = if (topic.assignments().nonEmpty) null else resolvedReplicationFactor

🤦‍♂ I see what you're saying now. I think @hachikuji's code snippet is the easiest to read so I'll change it to that. @ijuma my original line of thought was that it's easier to reason about the code if we only ever use resolvedX everywhere, hence the motivation for this change. That way I would never accidentally use the value of -1 in business logic code and it's easy to confirm by just looking at usages of topic.numPartitions and topic.replicationFactor.

agavra · 2019-05-21T23:44:23Z

Pinging +1'ers @colinhicks @omkreddy @gwenshap @rhauch 😄 thanks in advance! (cc @ijuma)

hachikuji · 2019-05-30T22:24:15Z

@agavra I may be missing something, but if we are not bumping the CreateTopic protocol version, how can the client know whether the default options are supported by the broker? In other words, if you try to use this API with an old broker, it seems like we'd get an INVALID_REPLICATION_FACTOR error, which is a little surprising. I'd expect to get an UNSUPPORTED_VERSION error instead.

core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala

agavra · 2019-05-30T22:45:14Z

@hachikuji - thanks for the review!

if we are not bumping the CreateTopic protocol version, how can the client know whether the default options are supported by the broker?

You are not missing something, this is something that I overlooked (not all errors are the same)!

hachikuji

Thanks for the updates. Left a few comments.

core/src/main/scala/kafka/server/AdminManager.scala

hachikuji · 2019-05-31T07:07:08Z

core/src/main/scala/kafka/server/AdminManager.scala

            val replicationFactor: java.lang.Short =
-              if (topic.replicationFactor == NO_REPLICATION_FACTOR) null else topic.replicationFactor
+              if (topic.replicationFactor == NO_REPLICATION_FACTOR) null else resolvedReplicationFactor


Hmm.. I think Ismael's point is that if topic.replicationFactor != NO_REPLICATION_FACTOR, then resolvedReplicationFactor == topic.replicationFactor, and this is equivalent to what we already had. It seems like this is what we're trying to do:

val replicationFactor: java.lang.Short = if (topic.assignments().nonEmpty) null else resolvedReplicationFactor

core/src/main/scala/kafka/admin/TopicCommand.scala

hachikuji · 2019-05-31T07:23:41Z

core/src/main/scala/kafka/admin/TopicCommand.scala

@@ -289,7 +294,7 @@ object TopicCommand extends Logging {
        if (topic.hasReplicaAssignment)
          adminZkClient.createTopicWithAssignment(topic.name, topic.configsToAdd, topic.replicaAssignment.get)
        else
-          adminZkClient.createTopic(topic.name, topic.partitions.get, topic.replicationFactor, topic.configsToAdd, topic.rackAwareMode)
+          adminZkClient.createTopic(topic.name, topic.partitions.get, topic.replicationFactor.get, topic.configsToAdd, topic.rackAwareMode)


The error may be a bit obscure if no --partitions or --replication-factor option was provided. Could we detect this case in the argument checks we have below?

I wasn't sure if it was worth adding bloat for the deprecated code, but I'm happy adding it back (I had it then removed it)

hachikuji

Thanks, looks good. Just a few more small comments.

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

core/src/main/scala/kafka/admin/TopicCommand.scala

core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala

hachikuji

Just one more comment, but LGTM overall.

hachikuji · 2019-06-03T18:51:55Z

core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala

+    createAndWaitTopic(new TopicCommandOptions(
+      Array("--topic", testTopicName)))
+
+    adminClient.listTopics().names().get().contains(testTopicName)


Can we add default assertions here. Maybe we can just use describeTopic like the other tests?

Oops i missed this comment! Addressing that now.

agavra · 2019-06-04T23:24:21Z

"retest this please"

…lient#createTopic. This commit makes three changes: - Adds a constructor for NewTopic(String, Optional<Integer>, Optional<Short>) which allows users to specify Optional.empty() for numPartitions or replicationFactor in order to use the broker default. - Changes AdminManager to accept -1 as valid options for replication factor and numPartitions (resolving to broker defaults). - Adds a dependency on scalaJava8Compat library to make it simpler to convert Scala Option to Java Optional

hachikuji

Thanks, LGTM. Merging to trunk!

…lient#createTopic (KIP-464) (apache#6728) This commit makes three changes: - Adds a constructor for NewTopic(String, Optional<Integer>, Optional<Short>) which allows users to specify Optional.empty() for numPartitions or replicationFactor in order to use the broker default. - Changes AdminManager to accept -1 as valid options for replication factor and numPartitions (resolving to broker defaults). - Makes --partitions and --replication-factor optional arguments when creating topics using kafka-topics.sh. - Adds a dependency on scalaJava8Compat library to make it simpler to convert Scala Option to Java Optional Reviewers: Ismael Juma <[email protected]>, Ryanne Dolan <[email protected]>, Jason Gustafson <[email protected]>

[KIP-464](https://cwiki.apache.org/confluence/display/KAFKA/KIP-464%3A+Defaults+for+AdminClient%23createTopic) (PR: #6728) made it possible to create topics without passing partition count and/or replica factor when using the admin client. We incorrectly disallowed this via #10457 while trying to ensure validation was consistent between ZK and the admin client (in this case the inconsistency was intentional). Fix this regression and add tests for the command lines in quick start (i.e. create topic and describe topic) to make sure it won't be broken in the future. Reviewers: Lee Dongjin <[email protected]>, Ismael Juma <[email protected]>

[KIP-464](https://cwiki.apache.org/confluence/display/KAFKA/KIP-464%3A+Defaults+for+AdminClient%23createTopic) (PR: apache/kafka#6728) made it possible to create topics without passing partition count and/or replica factor when using the admin client. We incorrectly disallowed this via apache/kafka#10457 while trying to ensure validation was consistent between ZK and the admin client (in this case the inconsistency was intentional). Fix this regression and add tests for the command lines in quick start (i.e. create topic and describe topic) to make sure it won't be broken in the future. Reviewers: Lee Dongjin <[email protected]>, Ismael Juma <[email protected]>

…he#11429) [KIP-464](https://cwiki.apache.org/confluence/display/KAFKA/KIP-464%3A+Defaults+for+AdminClient%23createTopic) (PR: apache#6728) made it possible to create topics without passing partition count and/or replica factor when using the admin client. We incorrectly disallowed this via apache#10457 while trying to ensure validation was consistent between ZK and the admin client (in this case the inconsistency was intentional). Fix this regression and add tests for the command lines in quick start (i.e. create topic and describe topic) to make sure it won't be broken in the future. Reviewers: Lee Dongjin <[email protected]>, Ismael Juma <[email protected]>

ryannedolan approved these changes May 14, 2019

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/admin/NewTopic.java Outdated Show resolved Hide resolved

ijuma reviewed May 15, 2019

View reviewed changes

hachikuji reviewed May 30, 2019

View reviewed changes

core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala Outdated Show resolved Hide resolved

hachikuji reviewed May 31, 2019

View reviewed changes

hachikuji approved these changes Jun 3, 2019

View reviewed changes

hachikuji approved these changes Jun 5, 2019

View reviewed changes

hachikuji merged commit 8e16158 into apache:trunk Jun 5, 2019

garyrussell mentioned this pull request Nov 27, 2019

Enhancement: Support Coercing replication-factor when default and Broker defines a policy spring-attic/spring-cloud-stream-binder-kafka#808

Closed

showuon mentioned this pull request Oct 24, 2021

KAFKA-13396: allow create topic without partition/replicaFactor #11429

Merged

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-8305: support default partitions & replication factor in AdminClient#createTopic #6728

KAFKA-8305: support default partitions & replication factor in AdminClient#createTopic #6728

agavra commented May 14, 2019 •

edited

Loading

ryannedolan left a comment

ijuma left a comment

ijuma May 15, 2019

agavra May 15, 2019

ijuma May 17, 2019

agavra May 19, 2019 •

edited

Loading

hachikuji May 31, 2019 •

edited

Loading

agavra May 31, 2019 •

edited

Loading

agavra commented May 21, 2019

hachikuji commented May 30, 2019 •

edited

Loading

agavra commented May 30, 2019

hachikuji left a comment

hachikuji May 31, 2019 •

edited

Loading

hachikuji May 31, 2019

agavra May 31, 2019

hachikuji left a comment

hachikuji left a comment

hachikuji Jun 3, 2019

agavra Jun 4, 2019

agavra commented Jun 4, 2019

hachikuji left a comment

KAFKA-8305: support default partitions & replication factor in AdminClient#createTopic #6728

KAFKA-8305: support default partitions & replication factor in AdminClient#createTopic #6728

Conversation

agavra commented May 14, 2019 • edited Loading

Description

Testing

Committer Checklist (excluded from commit message)

ryannedolan left a comment

Choose a reason for hiding this comment

ijuma left a comment

Choose a reason for hiding this comment

ijuma May 15, 2019

Choose a reason for hiding this comment

agavra May 15, 2019

Choose a reason for hiding this comment

ijuma May 17, 2019

Choose a reason for hiding this comment

agavra May 19, 2019 • edited Loading

Choose a reason for hiding this comment

hachikuji May 31, 2019 • edited Loading

Choose a reason for hiding this comment

agavra May 31, 2019 • edited Loading

Choose a reason for hiding this comment

agavra commented May 21, 2019

hachikuji commented May 30, 2019 • edited Loading

agavra commented May 30, 2019

hachikuji left a comment

Choose a reason for hiding this comment

hachikuji May 31, 2019 • edited Loading

Choose a reason for hiding this comment

hachikuji May 31, 2019

Choose a reason for hiding this comment

agavra May 31, 2019

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

hachikuji Jun 3, 2019

Choose a reason for hiding this comment

agavra Jun 4, 2019

Choose a reason for hiding this comment

agavra commented Jun 4, 2019

hachikuji left a comment

Choose a reason for hiding this comment

agavra commented May 14, 2019 •

edited

Loading

agavra May 19, 2019 •

edited

Loading

hachikuji May 31, 2019 •

edited

Loading

agavra May 31, 2019 •

edited

Loading

hachikuji commented May 30, 2019 •

edited

Loading

hachikuji May 31, 2019 •

edited

Loading