This is an automated email from the ASF dual-hosted git repository.
mxsm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 289bac6c6 [ISSUE #4266]InterruptedExceptions should never be
ignored[connectors] (#4314)
289bac6c6 is described below
commit 289bac6c6fd2927188b1920b479a3910d10565ab
Author: ruhshan <[email protected]>
AuthorDate: Wed Aug 2 15:47:34 2023 +0600
[ISSUE #4266]InterruptedExceptions should never be ignored[connectors]
(#4314)
* [ISSUE #4266]: Retore interrupted state for interrupted exception in open
function sinc connector
* [ISSUE #4266]: Retore interrupted state for interrupted exception in open
function source connector
* [ISSUE #4266]: Retore interrupted state for interrupted exception in
RocketMQSink connector
---
.../openfunction/sink/connector/OpenFunctionSinkConnector.java | 5 ++++-
.../openfunction/source/connector/OpenFunctionSourceConnector.java | 6 ++++--
.../connector/rocketmq/sink/connector/RocketMQSinkConnector.java | 5 +++++
3 files changed, 13 insertions(+), 3 deletions(-)
diff --git
a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/sink/connector/OpenFunctionSinkConnector.java
b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/sink/connector/OpenFunctionSinkConnector.java
index 80052f4ed..459f2d26b 100644
---
a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/sink/connector/OpenFunctionSinkConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/sink/connector/OpenFunctionSinkConnector.java
@@ -74,7 +74,10 @@ public class OpenFunctionSinkConnector implements Sink {
try {
queue.put(connectRecord);
} catch (InterruptedException e) {
- throw new RuntimeException(e);
+ Thread currentThread = Thread.currentThread();
+ log.warn("[OpenFunctionSinkConnector] Interrupting thread {}
due to exception {}",
+ currentThread.getName(), e.getMessage());
+ currentThread.interrupt();
}
}
}
diff --git
a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/connector/OpenFunctionSourceConnector.java
b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/connector/OpenFunctionSourceConnector.java
index c25948575..163c9b120 100644
---
a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/connector/OpenFunctionSourceConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/connector/OpenFunctionSourceConnector.java
@@ -88,8 +88,10 @@ public class OpenFunctionSourceConnector implements Source {
}
connectRecords.add(connectRecord);
} catch (InterruptedException e) {
- // nothing to do
- break;
+ Thread currentThread = Thread.currentThread();
+ log.warn("[OpenFunctionSourceConnector] Interrupting thread {}
due to exception {}",
+ currentThread.getName(), e.getMessage());
+ currentThread.interrupt();
}
}
return connectRecords;
diff --git
a/eventmesh-connectors/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/sink/connector/RocketMQSinkConnector.java
b/eventmesh-connectors/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/sink/connector/RocketMQSinkConnector.java
index f33e604e2..a65ecb26c 100644
---
a/eventmesh-connectors/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/sink/connector/RocketMQSinkConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/sink/connector/RocketMQSinkConnector.java
@@ -77,6 +77,11 @@ public class RocketMQSinkConnector implements Sink {
Message message = convertRecordToMessage(connectRecord);
try {
SendResult sendResult = producer.send(message);
+ } catch (InterruptedException e) {
+ Thread currentThread = Thread.currentThread();
+ log.warn("[RocketMQSinkConnector] Interrupting thread {} due
to exception {}",
+ currentThread.getName(), e.getMessage());
+ currentThread.interrupt();
} catch (Exception e) {
log.error("[RocketMQSinkConnector] sendResult has error : ",
e);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
