This is an automated email from the ASF dual-hosted git repository. rong pushed a commit to branch pipe-parallel-connector in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 876ca930ef27d30cdb2a4f756a3b8a797a8025a2 Author: Steve Yurong Su <[email protected]> AuthorDate: Mon Jun 12 01:37:28 2023 +0800 warn -> error --- .../db/pipe/connector/lagacy/IoTDBSyncConnectorImplV1_1.java | 8 ++++---- .../iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java | 4 ++-- .../iotdb/db/pipe/task/subtask/PipeConnectorSubtaskManager.java | 9 --------- 3 files changed, 6 insertions(+), 15 deletions(-) diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/lagacy/IoTDBSyncConnectorImplV1_1.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/lagacy/IoTDBSyncConnectorImplV1_1.java index 3099cf008cc..3ecfbe57246 100644 --- a/server/src/main/java/org/apache/iotdb/db/pipe/connector/lagacy/IoTDBSyncConnectorImplV1_1.java +++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/lagacy/IoTDBSyncConnectorImplV1_1.java @@ -26,7 +26,7 @@ import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException; import org.apache.iotdb.commons.pipe.config.PipeConfig; -import org.apache.iotdb.db.pipe.connector.IoTDBThriftConnectorClient; +import org.apache.iotdb.db.pipe.connector.v1.IoTDBThriftConnectorClient; import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; @@ -163,7 +163,7 @@ public class IoTDBSyncConnectorImplV1_1 implements PipeConnector { "IoTDBSyncConnectorV1_1 only support PipeInsertNodeInsertionEvent and PipeTabletInsertionEvent."); } } catch (TException e) { - LOGGER.error( + LOGGER.warn( "Network error when transfer tablet insertion event: {}.", tabletInsertionEvent, e); // the connection may be broken, try to reconnect by catching PipeConnectionException throw new PipeConnectionException( @@ -193,7 +193,7 @@ public class IoTDBSyncConnectorImplV1_1 implements PipeConnector { try { doTransfer((PipeTsFileInsertionEvent) tsFileInsertionEvent); } catch (TException e) { - LOGGER.error( + LOGGER.warn( "Network error when transfer tsFile insertion event: {}.", tsFileInsertionEvent, e); // The connection may be broken, try to reconnect by catching PipeConnectionException throw new PipeConnectionException("Network error when transfer tsFile insertion event.", e); @@ -244,7 +244,7 @@ public class IoTDBSyncConnectorImplV1_1 implements PipeConnector { } } } catch (TException e) { - LOGGER.error(String.format("Cannot send pipe data to receiver %s:%s.", ipAddress, port), e); + LOGGER.warn(String.format("Cannot send pipe data to receiver %s:%s.", ipAddress, port), e); throw new PipeConnectionException(e.getMessage(), e); } } diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java index 34ec4c2f63f..e863bad5dd9 100644 --- a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java +++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java @@ -132,7 +132,7 @@ public class IoTDBThriftConnectorV1 implements PipeConnector { "IoTDBThriftConnectorV1 only support PipeInsertNodeTabletInsertionEvent and PipeRawTabletInsertionEvent."); } } catch (TException e) { - LOGGER.error( + LOGGER.warn( "Network error when transfer tablet insertion event: {}.", tabletInsertionEvent, e); // the connection may be broken, try to reconnect by catching PipeConnectionException throw new PipeConnectionException( @@ -183,7 +183,7 @@ public class IoTDBThriftConnectorV1 implements PipeConnector { try { doTransfer((PipeTsFileInsertionEvent) tsFileInsertionEvent); } catch (TException e) { - LOGGER.error( + LOGGER.warn( "Network error when transfer tsfile insertion event: {}.", tsFileInsertionEvent, e); // the connection may be broken, try to reconnect by catching PipeConnectionException throw new PipeConnectionException( diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtaskManager.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtaskManager.java index fadb6a0184f..32740240388 100644 --- a/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtaskManager.java +++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtaskManager.java @@ -126,15 +126,6 @@ public class PipeConnectorSubtaskManager { attributeSortedString2SubtaskLifeCycleMap.get(attributeSortedString).stop(); } - public PipeConnectorSubtask getPipeConnectorSubtask(String attributeSortedString) { - if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) { - throw new PipeException( - "Failed to get PipeConnectorSubtask. No such subtask: " + attributeSortedString); - } - - return attributeSortedString2SubtaskLifeCycleMap.get(attributeSortedString).getSubtask(); - } - public BoundedBlockingPendingQueue<Event> getPipeConnectorPendingQueue( String attributeSortedString) { if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
