d8tltanc commented on a change in pull request #8683:
URL: https://github.com/apache/kafka/pull/8683#discussion_r441915398
##########
File path:
clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java
##########
@@ -300,30 +323,48 @@ public AuthenticationException
authenticationException(String id) {
* Resets the failure count for a node and sets the reconnect backoff to
the base
* value configured via reconnect.backoff.ms
*
- * @param nodeState The node state object to update
+ * @param nodeState nodeState The node state object to update
*/
private void resetReconnectBackoff(NodeConnectionState nodeState) {
nodeState.failedAttempts = 0;
- nodeState.reconnectBackoffMs = this.reconnectBackoffInitMs;
+ nodeState.reconnectBackoffMs = reconnectBackoff.term(0);
+ }
+
+ /**
+ * Resets the failure count for a node and sets the connection setup
timeout to the base
+ * value configured via socket.connection.setup.timeout.ms
+ *
+ * @param nodeState nodeState The node state object to update
+ */
+ private void resetConnectionSetupTimeout(NodeConnectionState nodeState) {
+ nodeState.failedConnectAttempts = 0;
+ nodeState.connectionSetupTimeoutMs = connectionSetupTimeout.term(0);
}
/**
- * Update the node reconnect backoff exponentially.
+ * Increment the failure counter, update the node reconnect backoff
exponentially,
+ * and record the current timestamp.
* The delay is reconnect.backoff.ms * 2**(failures - 1) * (+/- 20% random
jitter)
* Up to a (pre-jitter) maximum of reconnect.backoff.max.ms
*
* @param nodeState The node state object to update
*/
- private void updateReconnectBackoff(NodeConnectionState nodeState) {
- if (this.reconnectBackoffMaxMs > this.reconnectBackoffInitMs) {
- nodeState.failedAttempts += 1;
- double backoffExp = Math.min(nodeState.failedAttempts - 1,
this.reconnectBackoffMaxExp);
- double backoffFactor = Math.pow(RECONNECT_BACKOFF_EXP_BASE,
backoffExp);
- long reconnectBackoffMs = (long) (this.reconnectBackoffInitMs *
backoffFactor);
- // Actual backoff is randomized to avoid connection storms.
- double randomFactor = ThreadLocalRandom.current().nextDouble(0.8,
1.2);
- nodeState.reconnectBackoffMs = (long) (randomFactor *
reconnectBackoffMs);
- }
+ private void incrementReconnectBackoff(NodeConnectionState nodeState, long
now) {
+ nodeState.reconnectBackoffMs =
reconnectBackoff.term(nodeState.failedAttempts);
+ nodeState.failedAttempts++;
Review comment:
reconnectBackoff.term(0) will return the ${reconnect.backoff.ms} * 2 * 0
* jitter
reconnectBackoff.term(1) will return the ${reconnect.backoff.ms} * 2 * 1 *
jitter
The difference btw reconnect backoff and connection timeout here is that,
after the first failed attempts, the connection timeout will be the 1st term of
the randomized geometric sequence but the reconnect backoff will be the 0st
term of the randomized geometric sequence. So We should use (failedAttempts -
1) for fetching reconnect backoff and (failedAttemps) for fetching the
connection timeout.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
