Please note that this number (KIP-375) for "TopicCommand to use AdminClient" is discontinued due to a KIP number collision with KIP-375: Kafka Clients - make Metadata#TOPIC_EXPIRY_MS configurable and therefore not updated anymore.

Please continue the conversation on KIP-377: TopicCommand to use AdminClient.

...

Status

Current state: Under Discussion

...

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Currently kafka-topics.sh uses only direct Zookeeper connections which is not really desired compared to the AdminClient. This change would aim to add capability to the TopicCommand to be able to connect to a broker using the AdminClient.

...

• Allows clients in any language to administrate Kafka (Wire protocol is supported by any language)
• Provides public client for performing admin operations
• Ensures integration test code in other projects and clients maintains compatibility
• Prevents users from needing to use the Command classes and work around standard output and system exits
• Removing the need for admin scripts (kafka-configs.sh, kafka-topics.sh, etc) to talk directly to Zookeeper.
• Allows ZNodes to be completely locked down via ACLs
• Further hides the Zookeeper details of Kafka

Public Interfaces

Commandline Options

A few extra options will be added to kafka-configs.sh:

...

--command-config option will be also added. This would accept a file argument that points to the AdminClient configuration connection properties file (such as SSL, buffers, etc.).

val bootstrapServerOpt = parser.accepts("bootstrap-server", "REQUIRED: The Kafka servers to connect to, separated by commas, for instance "localhost:9091,localhost:9092". In case of providing this, a direct Zookeeper connection won't be required.")
                      .withRequiredArg
                      .describedAs("server to connect to")
                      .ofType(classOf[String])
val commandConfigOpt = parser.accepts("command-config", "Property file containing connection configs to be passed to Adminthe ClientAdminClient. " +
                      "This is used only with --bootstrap-server option for describing and altering broker configs.")
                      .withRequiredArg
                      .describedAs("command config property file")
                      .ofType(classOf[String])

...

Proposed Changes

The change proposed in this KIP is to add an extra option as stated above and to migrate create, delete, list and describe operations to use a broker connection. This would be a backward compatible change, meaning the zookeeper way would be still available until a further point in time but hopefully would deprecate it as part of this KIP.

Providing --bootstrap-server and --zookeeper together would result in an exception as they should be mutually exclusive.

Compatibility, Deprecation, And Migration Plan

This KIP won't implement topic config alternation as that is deprecated in the TopicCommand and should be done by kafka-configs.sh. The only alternation we allow is changing the partition number for topics.

No other existing behavior would be removed and the implementation would be done in a backward compatible way.

Test Plan

The existing tests will be run with the --bootstrap-server mode too. Additionally we can refactor some of the kafka-topics.sh usages in the smokes to use the AdminClient mode.

Rejected Alternatives

Protocol Changes

At an early stage of the KIP discussion it occured that there is a need for a protocol that would handle topic partition changes, such as increasing the partition number. It got rejected as a similar api, called CreatePartitions already exists and we don't need a new protocol.

For archiving purposes here is the protocol: kafka-topics.sh is also capable of altering the partition number, we need to add a protocol to implement this behavior. The simplest protocol we can give is where the request contains a list of topics with associated partition numbers and as a response we get the list of topics with the associated errors (or errors nulled out if it was successful).

AlterTopics Request (version: 0) => validate_only [topic_partition_change]
  validate_only => BOOLEAN
  topic_partition_change => topic_name target_partition_number
    topic_name => STRING
    target_partition_number => INT32


AlterTopics Response (version: 0) => throttle_time_ms [topic_partitionchange_result]
  throttle_time_ms => INT32
  topic_partitionchange_result => topic_name error_code error_message
    topic_name => STRING
    error_code => INT16
    error_message => NULLABLE_STRING

Authorization implications:

• From the authorization perspective using this protocol would require an ALTER operation on a Topic resource. Both currently available and therefore can be used.

...

Proposed Changes

The change proposed in this KIP is to add an extra option as stated above and to migrate create, delete, list and describe operations to use a broker connection. This would be a backward compatible change, meaning the zookeeper way would be still available until a further point in time but hopefully would deprecate it as part of this KIP.

Providing --bootstrap-server and --zookeeper together would result in an exception as they should be mutually exclusive.

Compatibility, Deprecation, And Migration Plan

This KIP won't implement topic config alternation as that is deprecated in the TopicCommand and should be done by kafka-configs.sh. The only alternation we allow is changing the partition number for topics.

No other existing behavior would be removed and the implementation would be done in a backward compatible way.

Test Plan

The existing tests will be run with the --bootstrap-server mode too. Additionally we can refactor some of the kafka-topics.sh usages in the smokes to use the AdminClient mode.

...