KAFKA-8305: support default partitions & replication factor in AdminClient#createTopic by agavra · Pull Request #6728 · apache/kafka

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!

ijuma

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

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

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.

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.

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

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

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

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.

hachikuji

Just one more comment, but LGTM overall.

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

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 <ismael@juma.me.uk>, Ryanne Dolan <ryannedolan@gmail.com>, Jason Gustafson <jason@confluent.io>

[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 <dongjin@apache.org>, Ismael Juma <ismael@juma.me.uk>

[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 <dongjin@apache.org>, Ismael Juma <ismael@juma.me.uk>

…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 <dongjin@apache.org>, Ismael Juma <ismael@juma.me.uk>

ryannedolan approved these changes May 14, 2019

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/clients/admin/NewTopic.java Outdated

ijuma reviewed May 15, 2019

View reviewed changes

hachikuji reviewed May 30, 2019

View reviewed changes

Comment thread core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala Outdated

hachikuji reviewed May 31, 2019

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/clients/admin/NewTopic.java Outdated

Comment thread core/src/main/scala/kafka/admin/TopicCommand.scala Outdated

Comment thread core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala Outdated

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

Conversation

agavra commented May 14, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Testing

Committer Checklist (excluded from commit message)

Uh oh!

ryannedolan left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

ijuma left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

ijuma May 15, 2019

Choose a reason for hiding this comment

Uh oh!

agavra May 15, 2019

Choose a reason for hiding this comment

Uh oh!

ijuma May 17, 2019

Choose a reason for hiding this comment

Uh oh!

agavra May 19, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji May 31, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

agavra May 31, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

agavra commented May 21, 2019

Uh oh!

hachikuji commented May 30, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

agavra commented May 30, 2019

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

hachikuji May 31, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

hachikuji May 31, 2019

Choose a reason for hiding this comment

Uh oh!

agavra May 31, 2019

Choose a reason for hiding this comment

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

hachikuji Jun 3, 2019

Choose a reason for hiding this comment

Uh oh!

agavra Jun 4, 2019

Choose a reason for hiding this comment

Uh oh!

agavra commented Jun 4, 2019

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

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