[jira] [Updated] (KAFKA-20393) Kafka client send to wong IP caused by stickyNode in TelemetrySender

["Yu Wang (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/KAFKA-20393?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Yu Wang updated KAFKA-20393:
----------------------------
    Description: 
In our case, we saw our Kafka consumer connected to a wrong Kafka pod because of IP reused by another Kafka cluster. (e.g. ClusterA broker 1 was replaced, then the IP of this broker used by ClusterB Broker 2)

After check the heap dump we found a stale telemetry stick node always cashed the old IP. This IP polluted the *connectionStates* of NetworkClient. Finally, when our consumer tried to get metadata, it fetches wrong metadata from the wrong Kafka cluster.

We think there are two places that cause this issue, both of them are in this method: 

[https://github.com/apache/kafka/blob/3.9.1/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java#L1312] 

1. When check the {*}canSendRequest{*}, it does not directly check the stickyNode, it uses the node id to get *connectionStates* and *selector channel* then do the check. In our case, at this moment, the there are new IP address in these two map and old address in stickyNode. Which makes the check pass and keeps the stale IP in stickyNode.
{code:java}
private long maybeUpdate(long now, Node node) {
String nodeConnectionId = node.idString();
if (canSendRequest(nodeConnectionId, now))
{ .... } {code}
 

 
{code:java}
private boolean canSendRequest(String node, long now) {
  return connectionStates.isReady(node, now) && selector.isChannelReady(node) &&
  inFlightRequests.canSendMore(node);
} {code}

2. When the stickyNode cannot pass the check of {*}canSendRequest{*}, it will set the stickyNode to null. But at this time the check of *connectionStates.canConnect* will still pass because the new IP is available in the connectionStates.

 
{code:java}
if (connectionStates.canConnect(nodeConnectionId, now)) {
  // We don't have a connection to this node right now, make one log.debug("Initialize connection to node {} for sending telemetry request", node);
  initiateConnect(node, now);
  return reconnectBackoffMs;
} {code}

This makes the code will got to {*}initiateConnect{*}. In this method, it will use the *host of the stickyNode* to initial the connection. Which pollute the host of connectionStates with the tale IP.

 
{code:java}
private void initiateConnect(Node node, long now) {
  String nodeConnectionId = node.idString();
  try {
    connectionStates.connecting(nodeConnectionId, now, node.host());
    InetAddress address = connectionStates.currentAddress(nodeConnectionId);
    log.debug("Initiating connection to node {} using address {}", node, address);
    selector.connect(nodeConnectionId,
    new InetSocketAddress(address, node.port()),
    this.socketSendBuffer,
    this.socketReceiveBuffer); {code}

Finally, when the Kafka client tried to connect the replaced broker again, it connected to a wrong IP, even worse, it get wrong metadata from another cluster.

  was:
In our case, we saw our Kafka consumer connected to a wrong Kafka pod because of IP reused by another Kafka cluster. (e.g. ClusterA broker 1 was replaced, then the IP of this broker used by ClusterB Broker 2)

After check the heap dump we found a stale telemetry stick node always cashed the old IP. This IP polluted the *connectionStates* of NetworkClient. Finally, when our consumer tried to get metadata, it fetches wrong metadata from the wrong Kafka cluster.

We think there are two places that cause this issue together:

1. When check the {*}canSendRequest{*}, it does not directly check the stickyNode, it uses the node id to get *connectionStates* and *selector channel* then do the check. In our case, at this moment, the there are new IP address in these two map and old address in stickyNode. Which makes the check pass and keeps the stale IP in stickyNode.
{color:#cf8e6d}private long {color}{color:#56a8f5}maybeUpdate{color}({color:#cf8e6d}long {color}now, Node node) {
String nodeConnectionId = node.idString();

{color:#cf8e6d}if {color}(canSendRequest(nodeConnectionId, now)) {
....
}
 
{color:#cf8e6d}private boolean {color}{color:#56a8f5}canSendRequest{color}(String node, {color:#cf8e6d}long {color}now) {
{color:#cf8e6d}return {color}{color:#c77dbb}connectionStates{color}.isReady(node, now) && {color:#c77dbb}selector{color}.isChannelReady(node) &&
{color:#c77dbb}inFlightRequests{color}.canSendMore(node);
}
2. When the stickyNode cannot pass the check of {*}canSendRequest{*}, it will set the stickyNode to null. But at this time the check of *connectionStates.canConnect* will still pass because the new IP is available in the connectionStates.
{color:#cf8e6d}if {color}({color:#c77dbb}connectionStates{color}.canConnect(nodeConnectionId, now)) {
{color:#7a7e85}// We don't have a connection to this node right now, make one
{color}{color:#7a7e85} {color}{color:#c77dbb}log{color}.debug({color:#6aab73}"Initialize connection to node {color}{color:#cf8e6d}{}{color}{color:#6aab73} for sending telemetry request"{color}, node);
initiateConnect(node, now);
{color:#cf8e6d}return {color}{color:#c77dbb}reconnectBackoffMs{color};
}
This makes the code will got to {*}initiateConnect{*}. In this method, it will use the *host of the stickyNode* to initial the connection. Which pollute the host of connectionStates with the tale IP.
{color:#cf8e6d}private void {color}{color:#56a8f5}initiateConnect{color}(Node node, {color:#cf8e6d}long {color}now) {
String nodeConnectionId = node.idString();
{color:#cf8e6d}try {color}{
{color:#c77dbb}connectionStates{color}.connecting(nodeConnectionId, now, node.host());
InetAddress address = {color:#c77dbb}connectionStates{color}.currentAddress(nodeConnectionId);
{color:#c77dbb}log{color}.debug({color:#6aab73}"Initiating connection to node {color}{color:#cf8e6d}{}{color}{color:#6aab73} using address {color}{color:#cf8e6d}{}{color}{color:#6aab73}"{color}, node, address);
{color:#c77dbb}selector{color}.connect(nodeConnectionId,
{color:#cf8e6d}new {color}InetSocketAddress(address, node.port()),
{color:#cf8e6d}this{color}.{color:#c77dbb}socketSendBuffer{color},
{color:#cf8e6d}this{color}.{color:#c77dbb}socketReceiveBuffer{color});
 
{color:#cf8e6d}private long {color}{color:#56a8f5}maybeUpdate{color}({color:#cf8e6d}long {color}now, Node node) {
String nodeConnectionId = node.idString();

{color:#cf8e6d}if {color}(canSendRequest(nodeConnectionId, now)) {
Optional<AbstractRequest.Builder<?>> requestOpt = {color:#c77dbb}clientTelemetrySender{color}.createRequest();

{color:#cf8e6d}if {color}(!requestOpt.isPresent())
{color:#cf8e6d}return {color}Long.{color:#c77dbb}MAX_VALUE{color};

AbstractRequest.Builder<?> request = requestOpt.get();
ClientRequest clientRequest = newClientRequest(nodeConnectionId, request, now, {color:#cf8e6d}true{color});
doSend(clientRequest, {color:#cf8e6d}true{color}, now);
{color:#cf8e6d}return {color}{color:#c77dbb}defaultRequestTimeoutMs{color};
} {color:#cf8e6d}else {color}{
{color:#7a7e85}// Per KIP-714, if we can't issue a request to this broker node, let's clear it out
{color}{color:#7a7e85} // and try another broker on the next loop.
{color}{color:#7a7e85} {color}{color:#c77dbb}stickyNode {color}= {color:#cf8e6d}null{color};
}

{color:#7a7e85}// If there's any connection establishment underway, wait until it completes. This prevents
{color}{color:#7a7e85} // the client from unnecessarily connecting to additional nodes while a previous connection
{color}{color:#7a7e85} // attempt has not been completed.
{color}{color:#7a7e85} {color}{color:#cf8e6d}if {color}(isAnyNodeConnecting())
{color:#cf8e6d}return {color}{color:#c77dbb}reconnectBackoffMs{color};

{color:#cf8e6d}if {color}({color:#c77dbb}connectionStates{color}.canConnect(nodeConnectionId, now)) {
{color:#7a7e85}// We don't have a connection to this node right now, make one
{color}{color:#7a7e85} {color}{color:#c77dbb}log{color}.debug({color:#6aab73}"Initialize connection to node {color}{color:#cf8e6d}{}{color}{color:#6aab73} for sending telemetry request"{color}, node);
initiateConnect(node, now);
{color:#cf8e6d}return {color}{color:#c77dbb}reconnectBackoffMs{color};
}

{color:#7a7e85}// In either case, we just need to wait for a network event to let us know the selected
{color}{color:#7a7e85} // connection might be usable again.
{color}{color:#7a7e85} {color}{color:#cf8e6d}return {color}Long.{color:#c77dbb}MAX_VALUE{color};
}
Finally, when the Kafka client tried to connect the replaced broker again, it connected to a wrong IP, even worse, it get wrong metadata from another cluster.


> Kafka client send to wong IP caused by stickyNode in TelemetrySender
> --------------------------------------------------------------------
>
>                 Key: KAFKA-20393
>                 URL: https://issues.apache.org/jira/browse/KAFKA-20393
>             Project: Kafka
>          Issue Type: Bug
>          Components: consumer, producer 
>    Affects Versions: 3.7.0, 3.9.1
>            Reporter: Yu Wang
>            Priority: Critical
>
> In our case, we saw our Kafka consumer connected to a wrong Kafka pod because of IP reused by another Kafka cluster. (e.g. ClusterA broker 1 was replaced, then the IP of this broker used by ClusterB Broker 2)
> After check the heap dump we found a stale telemetry stick node always cashed the old IP. This IP polluted the *connectionStates* of NetworkClient. Finally, when our consumer tried to get metadata, it fetches wrong metadata from the wrong Kafka cluster.
> We think there are two places that cause this issue, both of them are in this method: 
> [https://github.com/apache/kafka/blob/3.9.1/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java#L1312] 
> 1. When check the {*}canSendRequest{*}, it does not directly check the stickyNode, it uses the node id to get *connectionStates* and *selector channel* then do the check. In our case, at this moment, the there are new IP address in these two map and old address in stickyNode. Which makes the check pass and keeps the stale IP in stickyNode.
> {code:java}
> private long maybeUpdate(long now, Node node) {
> String nodeConnectionId = node.idString();
> if (canSendRequest(nodeConnectionId, now))
> { .... } {code}
>  
>  
> {code:java}
> private boolean canSendRequest(String node, long now) {
>   return connectionStates.isReady(node, now) && selector.isChannelReady(node) &&
>   inFlightRequests.canSendMore(node);
> } {code}
> 2. When the stickyNode cannot pass the check of {*}canSendRequest{*}, it will set the stickyNode to null. But at this time the check of *connectionStates.canConnect* will still pass because the new IP is available in the connectionStates.
>  
> {code:java}
> if (connectionStates.canConnect(nodeConnectionId, now)) {
>   // We don't have a connection to this node right now, make one log.debug("Initialize connection to node {} for sending telemetry request", node);
>   initiateConnect(node, now);
>   return reconnectBackoffMs;
> } {code}
> This makes the code will got to {*}initiateConnect{*}. In this method, it will use the *host of the stickyNode* to initial the connection. Which pollute the host of connectionStates with the tale IP.
>  
> {code:java}
> private void initiateConnect(Node node, long now) {
>   String nodeConnectionId = node.idString();
>   try {
>     connectionStates.connecting(nodeConnectionId, now, node.host());
>     InetAddress address = connectionStates.currentAddress(nodeConnectionId);
>     log.debug("Initiating connection to node {} using address {}", node, address);
>     selector.connect(nodeConnectionId,
>     new InetSocketAddress(address, node.port()),
>     this.socketSendBuffer,
>     this.socketReceiveBuffer); {code}
> Finally, when the Kafka client tried to connect the replaced broker again, it connected to a wrong IP, even worse, it get wrong metadata from another cluster.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)