rajinisivaram commented on code in PR #22187:
URL: https://github.com/apache/kafka/pull/22187#discussion_r3194789873
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1024,7 +1038,10 @@ private void
handleApiVersionsResponse(List<ClientResponse> responses,
InFlightRequest req, long now,
ApiVersionsResponse apiVersionsResponse) {
final String node = req.destination;
if (apiVersionsResponse.data().errorCode() != Errors.NONE.code()) {
- if (req.request.version() == 0 ||
apiVersionsResponse.data().errorCode() != Errors.UNSUPPORTED_VERSION.code()) {
+ if (metadataRecoveryStrategy ==
MetadataRecoveryStrategy.REBOOTSTRAP && apiVersionsResponse.data().errorCode()
== Errors.REBOOTSTRAP_REQUIRED.code()) {
+ log.info("Rebootstrap requested by server due to cluster
metadata mismatch.");
Review Comment:
I guess we don't have the broker-side info in the response to log what
didn't match. Maybe we could at least log the cluster id and node id that we
sent?
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -1529,10 +1529,22 @@ class KafkaApis(val requestChannel: RequestChannel,
apiVersionRequest.getErrorResponse(requestThrottleMs,
Errors.UNSUPPORTED_VERSION.exception)
} else if (!apiVersionRequest.isValid) {
apiVersionRequest.getErrorResponse(requestThrottleMs,
Errors.INVALID_REQUEST.exception)
+ } else if (clusterIdOrNodeIdIsInvalid(apiVersionRequest)) {
Review Comment:
Is there any risk that broker may not have initialized its cluster id during
start up when processing the first client request?
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -1529,10 +1529,22 @@ class KafkaApis(val requestChannel: RequestChannel,
apiVersionRequest.getErrorResponse(requestThrottleMs,
Errors.UNSUPPORTED_VERSION.exception)
} else if (!apiVersionRequest.isValid) {
apiVersionRequest.getErrorResponse(requestThrottleMs,
Errors.INVALID_REQUEST.exception)
+ } else if (clusterIdOrNodeIdIsInvalid(apiVersionRequest)) {
+ apiVersionRequest.getErrorResponse(requestThrottleMs,
Errors.REBOOTSTRAP_REQUIRED.exception)
} else {
apiVersionManager.apiVersionResponse(requestThrottleMs,
request.header.apiVersion() < 4)
}
}
+
+ // KIP-1242 checks the cluster ID and node ID in the request if provided
to ensure the
+ // client is connecting to the correct broker. If both are specified, they
must match
+ // the expected values for this broker.
+ def clusterIdOrNodeIdIsInvalid(apiVersionRequest: ApiVersionsRequest):
Boolean = {
+ apiVersionRequest.version >= 5 &&
+ apiVersionRequest.data.clusterId != null &&
+ (!apiVersionRequest.data.clusterId.equals(clusterId) ||
apiVersionRequest.data.nodeId != brokerId)
Review Comment:
nit: `apiVersionRequest.data.clusterId != clusterId`
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1107,6 +1124,19 @@ private void handleInitiateApiVersionRequests(long now) {
// not before ready.
this.connectionStates.checkingApiVersions(node);
ApiVersionsRequest.Builder apiVersionRequestBuilder =
entry.getValue();
+ // If we know the cluster ID and node ID we are connecting to,
we can include
+ // those details in the ApiVersions request for checking in
the broker,
+ // provided that the metadata recovery strategy is not NONE.
(KIP-1242)
+ if (metadataRecoveryStrategy != MetadataRecoveryStrategy.NONE
&& metadataClusterCheckEnable) {
+ String clusterId = this.metadataUpdater.clusterId();
+ int nodeId = Integer.parseInt(node);
+ // When connecting to coordinators, the client uses large
positive node ID
+ // values which do not match the target broker's node ID.
Exclude those.
+ if (clusterId != null && nodeId > 0 && nodeId <
Integer.MAX_VALUE / 2) {
Review Comment:
Can we define `Integer.MAX_VALUE / 2` as a constant with the comment about
coordinator's use of `Integer.MAX_VALUE - nodeId` which is the reason for using
half?
##########
clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java:
##########
@@ -297,6 +297,41 @@ defaultRequestTimeoutMs, connectionSetupTimeoutMsTest,
connectionSetupTimeoutMax
assertEquals(3, rebootstrapCount.get());
}
+ @Test
+ public void testMetadataClusterCheckFailureCausesRebootstrap() {
Review Comment:
This check is not testing `MetadataClusterCheckFailure`, it is simply
returns REBOOTSTRAP_REQUIRED in the response. We should rename this method to
reflect what it is actually doing. We also need a test to verify the new
broker-side validation - maybe we can add a test to `ApiVersionsRequestTest`.
Ideally, we would also want a test in `ClientRebootstrapTest` to validate
client-side handling of the response.
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1024,7 +1038,10 @@ private void
handleApiVersionsResponse(List<ClientResponse> responses,
InFlightRequest req, long now,
ApiVersionsResponse apiVersionsResponse) {
final String node = req.destination;
if (apiVersionsResponse.data().errorCode() != Errors.NONE.code()) {
- if (req.request.version() == 0 ||
apiVersionsResponse.data().errorCode() != Errors.UNSUPPORTED_VERSION.code()) {
+ if (metadataRecoveryStrategy ==
MetadataRecoveryStrategy.REBOOTSTRAP && apiVersionsResponse.data().errorCode()
== Errors.REBOOTSTRAP_REQUIRED.code()) {
Review Comment:
As @mingyen066 mentioned, we need to close all connections before
rebootstrapping. We should add a test as well to ensure connections are closed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
