Current state: Under Discussion
Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]
JIRA: here [Change the link from KAFKA-1 to your own ticket]
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
In some deployment environments such as Kubernetes, brokers may be assigned to worker nodes from an available pool. When a cluster rolls, it is possible that a broker changes its advertised hostname after a client has performed its initial bootstrap. As a result, a client may unwittingly use stale information to connect to a particular broker, only to connect to a different broker without realising. It is only later that the stale information becomes evident when things go badly wrong. Today, the only way to recover from this situation is to restart the client.
While it would be understandable to claim that this scenario is only possible because the cluster was not being orchestrated properly, the Kafka protocol does not pass any information from the client during session establishment which could spot this kind of inconsistency and automatically rebootstrap. It would also help with diagnosing situations where there is a mistake in the networking configuration for a Kafka cluster.
The Kafka protocol doesn't quite have sessions like many other protocols, but it does have an initial RPC used by a client when it connects to a broker which is ApiVersions . By adding optional ClusterId and NodeId information to the ApiVersions request, the receiving broker would be able to tell the client when it is attempting to make a misrouted connection.
This KIP introduces version 5 which adds ClusterId and NodeId to the request.
If the client is bootstrapping, it does not supply ClusterId or NodeId . After bootstrapping, during which is learns the information from its initial Metadata response, it supplies both.
The validation of ClusterId and NodeId is as follows:
ClusterId nor NodeId is specified, the request proceeds as normalClusterId or NodeId is specified, the request fails and the error code INVALID_REQUEST is returned.ClusterId and NodeId are specified and match the receiving broker, the request proceeds as normal.ClusterId is incorrect, the request fails and the error code INCONSISTENT_CLUSTER_ID is returned. This is a fatal error for the client.ClusterId is correct but the NodeId is incorrect, the request fails and the error code REBOOTSTRAP_REQUIRED is returned. This is a non-fatal error for the client which should rebootstrap.{
"apiKey": 18,
"type": "request",
"listeners": ["broker", "controller"],
"name": "ApiVersionsRequest",
// Versions 0 through 2 of ApiVersionsRequest are the same.
//
// Version 3 is the first flexible version and adds ClientSoftwareName and ClientSoftwareVersion.
//
// Version 4 fixes KAFKA-17011, which blocked SupportedFeatures.MinVersion in the response from being 0.
//
// Version 5 introduces ClusterId and NodeId (KIP-1242).
"validVersions": "0-5",
"flexibleVersions": "3+",
"fields": [
{ "name": "ClientSoftwareName", "type": "string", "versions": "3+",
"ignorable": true, "about": "The name of the client." },
{ "name": "ClientSoftwareVersion", "type": "string", "versions": "3+",
"ignorable": true, "about": "The version of the client." },
{ "name": "ClusterId", "type": "string", "versions": "5+", "nullableVersions": "5+", "ignorable": "true", "default": "null",
"about": "The cluster ID the client intends to connect to, if known." },
{ "name": "NodeId", "type": "int32", "versions": "5+", "default": -1, "ignorable": "true",
"about": "The broker ID the client intends to connect to, if known." }
]
} |
The response schema is unchanged. ApiVersions is now able to return two additional existing errors codes: REBOOTSTRAP_REQUIRED and INCONSISTENT_CLUSTER_ID .
There should be no negative impacts on existing users. Undiagnosed misrouted connections should be eliminated.
The KIP will be tested extensively using unit, integration and system tests. In particular, we should cause misrouted connections in a controlled way in order to check that the clients can rebootstrap automatically.
An alternative would be to introduce a new Connect RPC into the Kafka protocol, rather than extending ApiVersions . This would naturally extend to support new features such as multi-tenancy and namespaces in the future. We probably would not want to add additional context such as a namespace onto ApiVersions , so maybe we should bite the bullet now and add a new RPC. Part of the motivation of this KIP is to start the discussion in the community about the relative merits.