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 6fa6f914b [ISSUE #3830] InterruptedExceptions should not be ignored in
the code. (#4791)
6fa6f914b is described below
commit 6fa6f914b532c9d175a818701dc16088a10ffed7
Author: Jevin Jiang <[email protected]>
AuthorDate: Fri Mar 22 13:03:44 2024 +0800
[ISSUE #3830] InterruptedExceptions should not be ignored in the code.
(#4791)
* [ISSUE #3830] InterruptedExceptions should not be ignored in the code.
* [ISSUE #3830] fix code style
* [ISSUE #3830] delete Unneeded throw
* [ISSUE #3830] rethrown InterruptedException
* [ISSUE #3830] InterruptedException return new SendResult
---------
Co-authored-by: JiangShuJu <[email protected]>
---
.../eventmesh/storage/rocketmq/producer/ProducerImpl.java | 15 ++++++++++++++-
1 file changed, 14 insertions(+), 1 deletion(-)
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java
b/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java
index 681eda7a3..6789cdeb7 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java
@@ -81,6 +81,10 @@ public class ProducerImpl extends AbstractProducer {
messageId = sendResultRmq.getMsgId();
sendResult.setMessageId(messageId);
return sendResult;
+ } catch (InterruptedException e) {
+ log.error("Send message InterruptedException", e);
+ Thread.currentThread().interrupt(); // Restore interrupted status
+ return new SendResult();
} catch (Exception e) {
log.error(String.format("Send message Exception, %s", msg), e);
throw this.checkProducerException(msg.getTopic(), messageId, e);
@@ -94,6 +98,9 @@ public class ProducerImpl extends AbstractProducer {
supplySysProp(msg, cloudEvent);
try {
this.rocketmqProducer.sendOneway(msg);
+ } catch (InterruptedException e) {
+ log.error("Send message oneway InterruptedException", e);
+ Thread.currentThread().interrupt(); // Restore interrupted status
} catch (Exception e) {
log.error(String.format("Send message oneway Exception, %s", msg),
e);
throw this.checkProducerException(msg.getTopic(),
MessageClientIDSetter.getUniqID(msg), e);
@@ -104,9 +111,12 @@ public class ProducerImpl extends AbstractProducer {
this.checkProducerServiceState(this.rocketmqProducer.getDefaultMQProducerImpl());
org.apache.rocketmq.common.message.Message msg =
RocketMQMessageFactory.createWriter(Objects.requireNonNull(cloudEvent.getSubject())).writeBinary(cloudEvent);
- msg = supplySysProp(msg, cloudEvent);
+ supplySysProp(msg, cloudEvent);
try {
this.rocketmqProducer.send(msg, this.sendCallbackConvert(msg,
sendCallback));
+ } catch (InterruptedException e) {
+ log.error("Send message async InterruptedException", e);
+ Thread.currentThread().interrupt(); // Restore interrupted status
} catch (Exception e) {
log.error(String.format("Send message async Exception, %s", msg),
e);
throw this.checkProducerException(msg.getTopic(),
MessageClientIDSetter.getUniqID(msg), e);
@@ -134,6 +144,9 @@ public class ProducerImpl extends AbstractProducer {
try {
this.rocketmqProducer.send(msg, this.sendCallbackConvert(msg,
sendCallback));
+ } catch (InterruptedException e) {
+ log.error("Send message async InterruptedException", e);
+ Thread.currentThread().interrupt(); // Restore interrupted status
} catch (Exception e) {
log.error(String.format("Send message async Exception, %s", msg),
e);
throw this.checkProducerException(msg.getTopic(),
MessageClientIDSetter.getUniqID(msg), e);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
