Add create topics support #958

jlandersen · 2018-04-28T18:57:09Z

This is a first attempt at adding support for the create topics request. I have been wanting this for a while since I work on some "management" tools for Kafka. Most other Kafka clients in other languages also seem to rely on the "metadata" solution.

This supports specifying number of partitions and replication factor. Explicit replica assignments and config entries are left out for now.

A create topic request must be sent to the controller. To find the current controller, this also adds a version 1 metadata request, which includes the controller Id in the response.

There are a couple of things that are less-than-optimal and could use some input for improvement:

Added a "V2" suffix to resolve the conflict with current versions of createTopics(..) and loadMetadataForTopics(..).
KafkaClient includes these new functions, but the Admin constructor may be more appropriate for create topics, if this is intended to host more admin functionality in the future (would also resolve the createTopics(..) conflict on the client)?
The metadata structure returned from loadMetadataForTopics(..) could use a different structure. The first item in the array is an object with all the broker nodes. Second item is an object with a metadata property containing the topics. I've added the clusterMetadata property here, to avoid breaking the existing structure.
As a follow up to this, I've left the general handling of meta data load/refresh, but the new one could replace the existing one as I see it, so controller id is always available.

Let me know of anything we need to change before this can hopefully get merged in.

Once this is wrapped up, I have delete topics request ready next, and will then add support for config entries on creation of topics.

jlandersen · 2018-04-28T20:14:57Z

I am looking into the build issue with Kafka 0.9 (metadata version 1 was added in 0.10)

jlandersen · 2018-04-30T18:03:09Z

Updated PR with a few changes to handle clusters of version < 0.10.
Will now return an error if a controller is not found + tests are skipped for version 0.9 (so basically this feature is 0.10+).

Let me know what you think @hyperlink

hyperlink

Thanks for your contribution @jlandersen! 👍

It would be great if we could add an alias to createTopic in Admin. Also we should update the doc to reflect these changes.

hyperlink · 2018-05-03T16:00:13Z

lib/protocol/protocol.js

+    })
+    .word16bs('errorCode')
+    .tap((vars) => {
+      console.log(vars);


This WIP? Could we get rid of this console.log?

Definitely, will remove WIP parts

hyperlink · 2018-05-03T16:14:17Z

lib/kafkaClient.js

@@ -695,6 +726,29 @@ KafkaClient.prototype.clearCallbackQueue = function (socket, error) {
  delete this.cbqueue[socketId];
 };

+// Sends a version 1 metadata request which includes controller node in response
+KafkaClient.prototype.loadMetadataForTopicsV2 = function (topics, cb) {


We shouldn't continue to create these V{x} version of api calls. It creates unnecessary burden of managing kafka version protocol on the user:

Override Client loadMetadataForTopics

ensure the broker is ready and use getSupportedForRequestType to get the appropriate version of encoder/decoder (see example in sendRequest)

Makes sense - my one concern with this is is that the between api version 0 and >=1 the semantics have changed with regards to the topics parameter.

In version 0 an empty array returns all topics. In version >= 1, a null value (-1) is used to fetch all topics. An empty array returns no topic metadata [1]. This is potentially trouble for clients that use this. I guess this could be put under the next major version umbrella, thus allowing this breaking change. But there are also plenty of calls from inside the lib relying on this (and I think doing something like converting [] -> null would be a bit of a lie for clients).

A suggestion could be:

Override it, and make it work with getSupportedForRequestType, which includes the different behaviour depending on version, thus a breaking change similar to createTopics.

It is up to clients to be aware of this - basically if they use a version >= 0.10 null is now used for fetching all topics.

Allow internal use to force version 0 for now to avoid manipulating the input parameter in order to keep all existing functionality that works with < 0.10 and relies on a metadata call unaffected.

[1] (https://kafka.apache.org/protocol#The_Messages_Metadata).

Seems like an ugly change from a users perspective since it puts the burden on the user again to update the behavior of their app depending on the version they're connecting to and would make migrations painful. I think this is a case where we should keep kafka node's behavior consistent and abstract away the ugliness of the kafka protocol. Calling loadMetadataForTopics with empty array should always return all topics.

I always thought fetching all topic metadata seemed inefficient when we're just to get the broker metadata. So perhaps we can have new method called loadMetadata that will always pass [] for topic?

What do you think?

So the following changes:

Change existing loadMetaDataForTopics so that if the broker supports api version => 1, then convert an empty array to null value to fetch all topics

Add new loadMetadata that doesen't request for topics (by always passing [] for topic, I assume you mean fetch no topics in this case?). This can then be used in other cases where we just need broker metadata.

This sounds good to me - if you agree i'll try and get all the suggested changes in the PR in sometime this week.

sounds great @jlandersen thanks. 👍

hyperlink · 2018-05-03T16:16:54Z

lib/kafkaClient.js

+      return callback(error);
+    }
+
+    var controllerId = result[1].clusterMetadata.controllerId;


how often does the controllerId change? Would it be safe to cache this on the client so we can avoid doing a metadata request?

I have not been able to find documentation that verifies this - my impression is that the election process happens when starting the cluster and otherwise if the current elected broker crashes/becomes unavailable. We can cache it, and update metadata if the request ends up being sent to a broker that isn't controller?

@jlandersen caching it sounds good

hyperlink · 2018-05-03T16:19:32Z

lib/kafkaClient.js

+  broker.write(request);
+};
+
+KafkaClient.prototype.createTopicsV2 = function (topics, cb) {


I'm inclined to just call this createTopics. Planning to drop support node 4 so we're making major version change and this can be included in the things that "break".

Sounds good - I'll rename it to createTopics (and let the previous version remain on old client).

hyperlink · 2018-05-03T16:21:00Z

lib/kafkaClient.js

+
+KafkaClient.prototype.createTopicsV2 = function (topics, cb) {
+  var correlationId = this.nextId();
+  var request = protocol.encodeCreateTopicRequest(this.clientId, correlationId, topics, this.options.requestTimeout);


Could we move the request creation and correlationId a line down inside the handler ?

hyperlink · 2018-05-03T16:32:06Z

lib/protocol/protocol.js

+        }
+
+        const TIMEOUT_ERROR_CODE = 7;
+        if (vars.errorCode === TIMEOUT_ERROR_CODE) {


We should use the error map in the ERROR_CODE or just use the createGroupError function.

hyperlink · 2018-05-03T16:34:09Z

lib/protocol/protocol.js

+      this.buffer('topic', vars.topic);
+      vars.topic = vars.topic.toString();
+    })
+    .word16bs('errorCode')


WIP? We should handle this error code.

jlandersen · 2018-05-21T11:57:09Z

Bunch of changes are in:

Removed WIP parts (log, delete topics request encoder/decoder - will come in later PR with the full feature)
For version 1+ calls of metadata, [] values are sent as null to keep client behaviour of fetching all topics
Both createTopics and loadMetadataForTopics had their V2 suffix removed, so they replace the original implementations on Client.
I had to add support for existing calls to createTopic - if an array of strings is detected as input, the call is forwarded to original Client implementation. Lot's of tests depend on this, so rather than risking blowing it all up with this change, this will keep backwards compatibility for now.
loadMetadataForTopics on KafkaClient now uses getSupportedForRequestType (please check this bit, I added additional logic to make ensure broker is ready and the API version support is resolved).
Added loadMetadata function that refreshes metadata without topics.
Make sure of createErrorGroup when errors are detected in decoder (1)
Replaced README part on createTopics with new function documentation (we can also have both version described?)
Controller Id is now cached. It is updated any time updateMetadatas is called and the response has a controller id.

(1) for the createTopics decoder, I distinguish between two types of errors. Overall request errors and individual topic creation errors. Overall request errors are returned as an instance of Error (e.g. timeout or the request somehow got sent to a non-controller). Topic errors are returned as a normal result with the error message provided by Kafka (e.g. setting a replication factor higher than available brokers).

hyperlink · 2018-05-22T16:28:13Z

lib/kafkaClient.js

+    const correlationId = this.nextId();
+    const request = protocol.encodeCreateTopicRequest(this.clientId, correlationId, topics, this.options.requestTimeout);
+
+    this.queueCallback(broker.socket, correlationId, [protocol.decodeCreateTopicResponse, callback]);


Do we want to clear the cached controller metadata and call createTopics again when we get a NOT_CONTROLLER error?

We can. I am inclined to create an explicit error constructor function for this type of error so it's easier to handle.

I am not sure if createTopics is the best place to do this clearing and possibly retry logic. When more functionality is added that requires talking to the controller (such as deleting topics) those would be in the same situation. Should we add some function that can wrap requests that must go to the controller?

Good idea. In Client we have sendGroupRequest so possibly add a sendControllerRequest?

hyperlink · 2018-05-22T16:37:13Z

test/test.kafkaClient.js

+        }
+      ], (error, result) => {
+        should.not.exist(error);
+        result.should.be.empty;


can we also fetch the topic metadata and verify the partitions/replication factor for newly created topics?

Good point - added.

hyperlink · 2018-05-22T16:42:10Z

test/test.kafkaClient.js

+        this.skip();
+      }
+
+      sandbox = sinon.sandbox.create();


we don't need the sinon sandbox here and in the afterEach.

jlandersen · 2018-05-27T14:32:24Z

Updated with a sendControllerRequest method on KafkaClient. It wraps the provided callback with logic that handles a NotControllerError error. In this case the cached controller id is removed and the request is redone (based on the getController method that will then be forced to fetch latest metadata).

Let me know of any other issues/comments/suggestions, otherwise I consider this final for now :-)

hyperlink · 2018-06-01T16:21:47Z

lib/kafkaClient.js

+
+    callback = this.wrapControllerCheckIfNeeded(encoder, decoder, originalArgs, originalCallback);
+
+    this.queueCallback(controller.socket, correlationId, [decoder, callback]);


hi @jlandersen thanks for the update. Could you help me understand why wrapControllerCheckIfNeeded is needed? Would it not be simpler to just inline a function in place of the callback that checks if the error is an instance of NotControllerError and then set the controller Id to null and call sendControllerRequest again if it is?

The reason was to limit the retry to a single time (or potentially x times). If we add a function in place of the callback that always calls the sendControllerRequest on a NotControllerError, it could keep making requests if the error is returned repeatedly. I am not saying it can or will happen, but was rather to be safe. Let me know if you prefer otherwise and i'll put the change in!

hyperlink · 2018-06-06T16:29:03Z

I'll merge once conflicts are resolved. thanks @jlandersen

Adds support for sending a create topics request as a real protocol request. This supports specifying number of partitions and replication factor. Explicit replica assignments and config entries are left out for now. A create topic request must be sent to the controller. To find the current controller, this also adds a version 1 metadata request, which includes the controller Id in the response.

Error message changes a bit depending on version. This relaxes the assertion a bit.

metadata v1 is not supported until 0.10

Happens when cluster is is running version less than 0.10

The order resulted in isInternal being used as the number of partitions during decoding.

- Add createTopics alias on admin client - Remove V2 suffix, replacing old versions of createTopics and loadMetadataForTopics on KafkaClient - Add loadMetadata function - Use supported API to resolve metadata request version

Now fetches metadata again to verify topics were created with requested partition count and replication factor.

Wraps the provided callback with retry logic if NotController error was returned. In this case the cached controller is cleared and it is fetched again by getting the latest metadata.

jlandersen · 2018-06-10T13:53:30Z

Rebased on master and all checks out! Thanks! Looking forward to adding more things with the controller request in place!

hyperlink · 2018-09-05T18:35:25Z

Published as 3.0.0

hyperlink reviewed May 3, 2018

View reviewed changes

jlandersen closed this May 21, 2018

jlandersen reopened this May 21, 2018

hyperlink reviewed May 22, 2018

View reviewed changes

jlandersen force-pushed the create-topic branch 2 times, most recently from caeafc5 to 4483cd3 Compare May 27, 2018 14:20

hyperlink reviewed Jun 1, 2018

View reviewed changes

jlandersen added 13 commits June 10, 2018 10:44

Handle different error message in kafka versions

b7e53c5

Error message changes a bit depending on version. This relaxes the assertion a bit.

Remove metadata api support version

de03d2d

metadata v1 is not supported until 0.10

Return error if controller cannot be found

1c46c1b

Happens when cluster is is running version less than 0.10

Parse partition metadata correctly for v1

d91091b

The order resulted in isInternal being used as the number of partitions during decoding.

Address PR feedback

d904075

- Add createTopics alias on admin client - Remove V2 suffix, replacing old versions of createTopics and loadMetadataForTopics on KafkaClient - Add loadMetadata function - Use supported API to resolve metadata request version

Encode null value properly

038c903

Forward old createTopics calls to previous impl

2916d70

Update README with new createTopics function

0ec81c8

Update cached metadata in getController

59052d5

Remove unused sandbox

d039f44

Add additional assertion of createTopic test

911e3e2

Now fetches metadata again to verify topics were created with requested partition count and replication factor.

Add sendControllerRequest method

ac6d07e

Wraps the provided callback with retry logic if NotController error was returned. In this case the cached controller is cleared and it is fetched again by getting the latest metadata.

jlandersen force-pushed the create-topic branch from 4483cd3 to ac6d07e Compare June 10, 2018 13:43

hyperlink approved these changes Jun 10, 2018

View reviewed changes

hyperlink merged commit a3653ba into SOHU-Co:master Jun 10, 2018

jlandersen deleted the create-topic branch June 11, 2018 18:13

hyperlink mentioned this pull request Jun 28, 2018

What is difference btw Highlevel producer and producer ? #275

Closed

hyperlink mentioned this pull request Sep 13, 2018

Delete a topic (feature request) #237

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add create topics support #958

Add create topics support #958

jlandersen commented Apr 28, 2018

jlandersen commented Apr 28, 2018

jlandersen commented Apr 30, 2018 •

edited

Loading

hyperlink left a comment

hyperlink May 3, 2018

jlandersen May 5, 2018

hyperlink May 3, 2018

jlandersen May 5, 2018 •

edited

Loading

hyperlink May 14, 2018

jlandersen May 15, 2018

hyperlink May 15, 2018

hyperlink May 3, 2018

jlandersen May 5, 2018

hyperlink May 14, 2018

hyperlink May 3, 2018

jlandersen May 5, 2018

hyperlink May 3, 2018

hyperlink May 3, 2018

hyperlink May 3, 2018

jlandersen commented May 21, 2018 •

edited

Loading

hyperlink May 22, 2018 •

edited

Loading

jlandersen May 22, 2018

hyperlink May 22, 2018

hyperlink May 22, 2018

jlandersen May 22, 2018

hyperlink May 22, 2018

jlandersen commented May 27, 2018 •

edited

Loading

hyperlink Jun 1, 2018

jlandersen Jun 1, 2018 •

edited

Loading

hyperlink commented Jun 6, 2018

jlandersen commented Jun 10, 2018

hyperlink commented Sep 5, 2018


		callback = this.wrapControllerCheckIfNeeded(encoder, decoder, originalArgs, originalCallback);

		this.queueCallback(controller.socket, correlationId, [decoder, callback]);

Add create topics support #958

Add create topics support #958

Conversation

jlandersen commented Apr 28, 2018

jlandersen commented Apr 28, 2018

jlandersen commented Apr 30, 2018 • edited Loading

hyperlink 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

jlandersen May 5, 2018 • 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

jlandersen commented May 21, 2018 • edited Loading

hyperlink May 22, 2018 • 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

jlandersen commented May 27, 2018 • edited Loading

Choose a reason for hiding this comment

jlandersen Jun 1, 2018 • edited Loading

Choose a reason for hiding this comment

hyperlink commented Jun 6, 2018

jlandersen commented Jun 10, 2018

hyperlink commented Sep 5, 2018

jlandersen commented Apr 30, 2018 •

edited

Loading

jlandersen May 5, 2018 •

edited

Loading

jlandersen commented May 21, 2018 •

edited

Loading

hyperlink May 22, 2018 •

edited

Loading

jlandersen commented May 27, 2018 •

edited

Loading

jlandersen Jun 1, 2018 •

edited

Loading