Pipe: Avoid throwing null pointer during Close process (apache#16391)

* Pipe: Avoid throwing null pointer during Close process

* fix

* fix

Author: luoluoyuyu

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java

@@ -95,6 +95,7 @@ public Pair<IoTDBSyncClient, Boolean> getClient(final String deviceId) {
             && endPoint != null
             && endPoint2ClientAndStatus.containsKey(endPoint)
             && Boolean.TRUE.equals(endPoint2ClientAndStatus.get(endPoint).getRight())
+            && endPoint2ClientAndStatus.get(endPoint).getLeft() != null
         ? endPoint2ClientAndStatus.get(endPoint)
         : getClient();
   }
@@ -104,6 +105,7 @@ public Pair<IoTDBSyncClient, Boolean> getClient(final TEndPoint endPoint) {
             && endPoint != null
             && endPoint2ClientAndStatus.containsKey(endPoint)
             && Boolean.TRUE.equals(endPoint2ClientAndStatus.get(endPoint).getRight())
+            && endPoint2ClientAndStatus.get(endPoint).getLeft() != null
         ? endPoint2ClientAndStatus.get(endPoint)
         : getClient();
   }

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManager.java

@@ -121,7 +121,7 @@ public void checkClientStatusAndTryReconstructIfNecessary() {
       // Check whether any clients are available, if any client is available, return directly
       for (final Pair<IoTDBSyncClient, Boolean> clientAndStatus :
           endPoint2ClientAndStatus.values()) {
-        if (Boolean.TRUE.equals(clientAndStatus.getRight())) {
+        if (Boolean.TRUE.equals(clientAndStatus.getRight()) && clientAndStatus.getLeft() != null) {
           return;
         }
       }
@@ -130,7 +130,7 @@ public void checkClientStatusAndTryReconstructIfNecessary() {
     // Reconstruct all dead clients
     for (final Map.Entry<TEndPoint, Pair<IoTDBSyncClient, Boolean>> entry :
         endPoint2ClientAndStatus.entrySet()) {
-      if (Boolean.TRUE.equals(entry.getValue().getRight())) {
+      if (Boolean.TRUE.equals(entry.getValue().getRight()) && entry.getValue().getLeft() != null) {
         continue;
       }
 
@@ -139,7 +139,7 @@ public void checkClientStatusAndTryReconstructIfNecessary() {
 
     // Check whether any clients are available
     for (final Pair<IoTDBSyncClient, Boolean> clientAndStatus : endPoint2ClientAndStatus.values()) {
-      if (Boolean.TRUE.equals(clientAndStatus.getRight())) {
+      if (Boolean.TRUE.equals(clientAndStatus.getRight()) && clientAndStatus.getLeft() != null) {
         lastCheckClientStatusTimestamp = System.currentTimeMillis();
         return;
       }
@@ -306,7 +306,6 @@ public void close() {
       try {
         if (clientAndStatus.getLeft() != null) {
           clientAndStatus.getLeft().close();
-          clientAndStatus.setLeft(null);
         }
         LOGGER.info("Client {}:{} closed.", endPoint.getIp(), endPoint.getPort());
       } catch (Exception e) {
@@ -337,7 +336,7 @@ public Pair<IoTDBSyncClient, Boolean> getClient() {
         final int clientIndex = (int) (currentClientIndex++ % clientSize);
         final Pair<IoTDBSyncClient, Boolean> clientAndStatus =
             endPoint2ClientAndStatus.get(endPointList.get(clientIndex));
-        if (Boolean.TRUE.equals(clientAndStatus.getRight())) {
+        if (Boolean.TRUE.equals(clientAndStatus.getRight()) && clientAndStatus.getLeft() != null) {
           return clientAndStatus;
         }
       }
@@ -354,7 +353,7 @@ public Pair<IoTDBSyncClient, Boolean> getClient() {
       final int clientIndex = (int) (Math.random() * clientSize);
       final Pair<IoTDBSyncClient, Boolean> clientAndStatus =
           endPoint2ClientAndStatus.get(endPointList.get(clientIndex));
-      if (Boolean.TRUE.equals(clientAndStatus.getRight())) {
+      if (Boolean.TRUE.equals(clientAndStatus.getRight()) && clientAndStatus.getLeft() != null) {
         return clientAndStatus;
       }
 
@@ -363,7 +362,8 @@ public Pair<IoTDBSyncClient, Boolean> getClient() {
         final int nextClientIndex = (clientIndex + tryCount + 1) % clientSize;
         final Pair<IoTDBSyncClient, Boolean> nextClientAndStatus =
             endPoint2ClientAndStatus.get(endPointList.get(nextClientIndex));
-        if (Boolean.TRUE.equals(nextClientAndStatus.getRight())) {
+        if (Boolean.TRUE.equals(nextClientAndStatus.getRight())
+            && clientAndStatus.getLeft() != null) {
           return nextClientAndStatus;
         }
       }
@@ -380,7 +380,7 @@ public Pair<IoTDBSyncClient, Boolean> getClient() {
       for (final TEndPoint endPoint : endPointList) {
         final Pair<IoTDBSyncClient, Boolean> clientAndStatus =
             endPoint2ClientAndStatus.get(endPoint);
-        if (Boolean.TRUE.equals(clientAndStatus.getRight())) {
+        if (Boolean.TRUE.equals(clientAndStatus.getRight()) && clientAndStatus.getLeft() != null) {
           return clientAndStatus;
         }
       }