unknowntpo commented on code in PR #23673:
URL: https://github.com/apache/kafka/pull/23673#discussion_r4281803407
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -135,17 +128,13 @@ private enum State {
private final MetadataRecoveryStrategy metadataRecoveryStrategy;
- /* Whether to send the cluster ID and node ID on ApiVersions RPC for
checking by the broker */
- private final boolean metadataClusterCheckEnable;
-
private final Time time;
- /**
- * True if we should send an ApiVersionRequest when first connecting to a
broker.
- */
+ private final ApiVersions apiVersions;
+
private final boolean discoverBrokerVersions;
- private final ApiVersions apiVersions;
+ private final boolean metadataClusterCheckEnable;
private final Map<String, ApiVersionsRequest.Builder>
nodesNeedingApiVersionsFetch = new HashMap<>();
Review Comment:
There's another way to refactor this part; leaving the idea here for
discussion.
`CHECKING_API_VERSIONS` is already a `ConnectionState`, but two of its
neighbours are not. "Connected, waiting for the channel to become ready before
sending ApiVersions" and "rejected, waiting to resend with a lower version" are
both expressed today as "the node is in `nodesNeedingApiVersionsFetch`", which
lives in `NetworkClient` next to `connectionStates` and has to be cleared by
hand in `processDisconnection` and `close(nodeId)`.
Instead of a separate map, `NodeConnectionState` could carry a
`pendingApiVersionsVersion` field:
- `ClusterConnectionStates` gets `awaitingApiVersions(node, maxVersion)`
(called from `handleConnections`), `retryApiVersions(node, fallbackVersion)`
(the RETRY case), and `nodesAwaitingApiVersions()`.
- `checkingApiVersions(node)` returns the pending version as part of the
transition, so the state change and the hand-off cannot get out of step.
- `disconnected`, `ready`, `remove` and a reconnect clear it, so the two
manual removals in `NetworkClient` go away.
`handleInitiateApiVersionRequests` then reads:
```java
for (String node : connectionStates.nodesAwaitingApiVersions()) {
if (!selector.isChannelReady(node) ||
!inFlightRequests.canSendMore(node))
continue;
short maxVersion = connectionStates.checkingApiVersions(node);
ApiVersionsRequest.Builder request = new
ApiVersionsRequest.Builder(maxVersion);
prepareApiVersionsRequest(node, request);
doSend(newClientRequest(node, request, now, true), true, now);
}
```
`NetworkClient` keeps the readiness check, the send and the response
handling, so this is a small move rather than a new abstraction, and the
sub-state becomes testable in `ClusterConnectionStatesTest` without
`MockSelector`. Happy to help if you want to go this way.
--
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]
