(iotdb) branch master updated: Pipe: Tolerate the error of listening queues in cluster restart (#13434)

[rong]

This is an automated email from the ASF dual-hosted git repository.

rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new 0be6fdbe545 Pipe: Tolerate the error of listening queues in cluster 
restart (#13434)
0be6fdbe545 is described below

commit 0be6fdbe5458988c0a0d91a3dfe986fc110245c5
Author: Caideyipi <87789683+caidey...@users.noreply.github.com>
AuthorDate: Mon Sep 9 12:08:05 2024 +0800

    Pipe: Tolerate the error of listening queues in cluster restart (#13434)
---
 .../pipe/extractor/ConfigRegionListeningQueue.java    | 19 +++++++++++++++----
 .../schemaregion/SchemaRegionListeningQueue.java      |  4 ++--
 2 files changed, 17 insertions(+), 6 deletions(-)

diff --git 
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/ConfigRegionListeningQueue.java
 
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/ConfigRegionListeningQueue.java
index 4c908667a9f..8fe2dce73a8 100644
--- 
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/ConfigRegionListeningQueue.java
+++ 
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/ConfigRegionListeningQueue.java
@@ -155,14 +155,25 @@ public class ConfigRegionListeningQueue extends 
AbstractPipeListeningQueue
   /////////////////////////////// Snapshot ///////////////////////////////
 
   @Override
-  public synchronized boolean processTakeSnapshot(final File snapshotDir)
-      throws TException, IOException {
-    return super.serializeToFile(new File(snapshotDir, SNAPSHOT_FILE_NAME));
+  public synchronized boolean processTakeSnapshot(final File snapshotDir) 
throws IOException {
+    try {
+      return super.serializeToFile(new File(snapshotDir, SNAPSHOT_FILE_NAME));
+    } catch (final IOException e) {
+      throw e;
+    } catch (final Exception e) {
+      throw new IOException(e);
+    }
   }
 
   @Override
   public synchronized void processLoadSnapshot(final File snapshotDir)
       throws TException, IOException {
-    super.deserializeFromFile(new File(snapshotDir, SNAPSHOT_FILE_NAME));
+    try {
+      super.deserializeFromFile(new File(snapshotDir, SNAPSHOT_FILE_NAME));
+    } catch (final IOException e) {
+      throw e;
+    } catch (final Exception e) {
+      throw new IOException(e);
+    }
   }
 }
diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/SchemaRegionListeningQueue.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/SchemaRegionListeningQueue.java
index 70ad8b6df24..42b3d685f73 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/SchemaRegionListeningQueue.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/SchemaRegionListeningQueue.java
@@ -103,7 +103,7 @@ public class SchemaRegionListeningQueue extends 
AbstractPipeListeningQueue {
   public synchronized boolean createSnapshot(final File snapshotDir) {
     try {
       return super.serializeToFile(new File(snapshotDir, SNAPSHOT_FILE_NAME));
-    } catch (final IOException e) {
+    } catch (final Exception e) {
       LOGGER.warn("Take snapshot error: {}", e.getMessage());
       return false;
     }
@@ -112,7 +112,7 @@ public class SchemaRegionListeningQueue extends 
AbstractPipeListeningQueue {
   public synchronized void loadSnapshot(final File snapshotDir) {
     try {
       super.deserializeFromFile(new File(snapshotDir, SNAPSHOT_FILE_NAME));
-    } catch (final IOException e) {
+    } catch (final Exception e) {
       LOGGER.error("Failed to load snapshot {}", e.getMessage());
     }
   }