MINOR: Fix some spelling corrections in comments
Reviewers: Manikumar Reddy
@@ -869,7 +869,7 @@ public class NetworkClient implements KafkaClient {
*/
private void handleConnections() {
for (String node : this.selector.connected()) {
- // We are now connected. Node that we might not still be able to send requests. For instance,
+ // We are now connected. Note that we might not still be able to send requests. For instance,
// if SSL is enabled, the SSL handshake happens after the connection is established.
// Therefore, it is still necessary to check isChannelReady before attempting to send on this
// connection.
@@ -987,7 +987,7 @@ public class NetworkClient implements KafkaClient {
Cluster cluster = metadata.fetch();
// 'processDisconnection' generates warnings for misconfigured bootstrap server configuration
// resulting in 'Connection Refused' and misconfigured security resulting in authentication failures.
- // The warning below handles the case where connection to a broker was established, but was disconnected
+ // The warning below handles the case where a connection to a broker was established, but was disconnected
// before metadata could be obtained.
if (cluster.isBootstrapConfigured()) {
int nodeId = Integer.parseInt(destination);
@@ -1008,8 +1008,9 @@ public class NetworkClient implements KafkaClient {
@Override
public void handleCompletedMetadataResponse(RequestHeader requestHeader, long now, MetadataResponse response) {
- // If any partition has leader with missing listeners, log a few for diagnosing broker configuration
- // issues. This could be a transient issue if listeners were added dynamically to brokers.
+ // If any partition has leader with missing listeners, log up to ten of these partitions
+ // for diagnosing broker configuration issues.
+ // This could be a transient issue if listeners were added dynamically to brokers.
List<TopicPartition> missingListenerPartitions = response.topicMetadata().stream().flatMap(topicMetadata ->
topicMetadata.partitionMetadata().stream()
.filter(partitionMetadata -> partitionMetadata.error() == Errors.LISTENER_NOT_FOUND)
@@ -1021,12 +1022,12 @@ public class NetworkClient implements KafkaClient {
count, missingListenerPartitions.subList(0, Math.min(10, count)));
}
- // check if any topics metadata failed to get updated
+ // Check if any topic's metadata failed to get updated
Map<String, Errors> errors = response.errors();
if (!errors.isEmpty())
log.warn("Error while fetching metadata with correlation id {} : {}", requestHeader.correlationId(), errors);
- // don't update the cluster if there are no valid nodes...the topic we want may still be in the process of being
+ // Don't update the cluster if there are no valid nodes...the topic we want may still be in the process of being
// created which means we will get errors and no nodes until it exists
if (response.brokers().isEmpty()) {
log.trace("Ignoring empty metadata response with correlation id {}.", requestHeader.correlationId());
@@ -1085,7 +1086,7 @@ public class NetworkClient implements KafkaClient {
}
if (connectionStates.canConnect(nodeConnectionId, now)) {
- // we don't have a connection to this node right now, make one
+ // We don't have a connection to this node right now, make one
log.debug("Initialize connection to node {} for sending metadata request", node);
initiateConnect(node, now);
return reconnectBackoffMs;
