Skip to content

[ISSUE #3830] InterruptedExceptions should not be ignored in the code. #4791

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Mar 22, 2024
Merged

[ISSUE #3830] InterruptedExceptions should not be ignored in the code. #4791

Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
88dd059
[ISSUE #3830] InterruptedExceptions should not be ignored in the code.
Mar 17, 2024
5fa0c10
[ISSUE #3830] fix code style
Mar 18, 2024
58651cf
[ISSUE #3830] delete Unneeded throw
jevinjiang Mar 19, 2024
2bc1de4
[ISSUE #3830] rethrown InterruptedException
jevinjiang Mar 19, 2024
ab06b18
[ISSUE #3830] InterruptedException return new SendResult
jevinjiang Mar 20, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 29 additions & 1 deletion ...e-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,13 @@
messageId = sendResultRmq.getMsgId();
sendResult.setMessageId(messageId);
return sendResult;
} catch (InterruptedException e) {
log.error("Send message InterruptedException", e);
Thread.currentThread().interrupt(); // Restore interrupted status
throw this.checkProducerException(msg.getTopic(), messageId, e);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since you have chosen to re-interrupt, is there no need to throw an exception again? Moreover, the throws are all non-InterruptedExceptions.

既然您选择了重新中断是不是就没必要再抛异常了？更何况抛出都是一些非InterruptedException。

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If I don't throw an exception, I will return a SendResult, so I chose to throw a StorageRuntimeException. Is there any better way? I can continue to modify it. 不抛出异常的话,就要返回一个SendResult,所以我选择了抛出一个StorageRuntimeException, 还有什么更好的做法没,我可以继续修改

Copy link
Member

@pandaapo pandaapo Mar 19, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think you can either return an empty SendResult, or you can use the other way of handling InterruptedException mentioned in the issue, "InterruptedExceptions should either be rethrown - immediately...".

我觉得您可以返回空的SendResult，或者采用issue中提到的另一中处理InterruptedException的方式“直接抛出中断异常”。

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay, it's done. Thank you for your guidance in helping me complete the first issue。 已经修改完毕，感谢您帮我完成第一个issue。

} catch (ThreadDeath e) {
log.error("Send message ThreadDeath", e);
throw this.checkProducerException(msg.getTopic(), messageId, e);

Check warning on line 90 in eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java

View check run for this annotation

Codecov / codecov/patch

eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java#L84-L90 

Added lines #L84 - L90 were not covered by tests
} catch (Exception e) {
log.error(String.format("Send message Exception, %s", msg), e);
throw this.checkProducerException(msg.getTopic(), messageId, e);
Expand All @@ -94,6 +101,13 @@
supplySysProp(msg, cloudEvent);
try {
this.rocketmqProducer.sendOneway(msg);
} catch (InterruptedException e) {
log.error("Send message oneway InterruptedException", e);
Thread.currentThread().interrupt(); // Restore interrupted status
throw this.checkProducerException(msg.getTopic(), MessageClientIDSetter.getUniqID(msg), e);
} catch (ThreadDeath e) {
log.error("Send message oneway ThreadDeath", e);
throw this.checkProducerException(msg.getTopic(), MessageClientIDSetter.getUniqID(msg), e);

Check warning on line 110 in eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java

View check run for this annotation

Codecov / codecov/patch

eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java#L104-L110 

Added lines #L104 - L110 were not covered by tests
} catch (Exception e) {
log.error(String.format("Send message oneway Exception, %s", msg), e);
throw this.checkProducerException(msg.getTopic(), MessageClientIDSetter.getUniqID(msg), e);
Expand All @@ -104,9 +118,16 @@
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);

Check warning on line 121 in eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java

View check run for this annotation

Codecov / codecov/patch

eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java#L121 

Added line #L121 was not covered by tests
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
throw this.checkProducerException(msg.getTopic(), MessageClientIDSetter.getUniqID(msg), e);
} catch (ThreadDeath e) {
log.error("Send message async ThreadDeath", e);
throw this.checkProducerException(msg.getTopic(), MessageClientIDSetter.getUniqID(msg), e);

Check warning on line 130 in eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java

View check run for this annotation

Codecov / codecov/patch

eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java#L124-L130 

Added lines #L124 - L130 were not covered by tests
} catch (Exception e) {
log.error(String.format("Send message async Exception, %s", msg), e);
throw this.checkProducerException(msg.getTopic(), MessageClientIDSetter.getUniqID(msg), e);
Expand Down Expand Up @@ -134,6 +155,13 @@

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
throw this.checkProducerException(msg.getTopic(), MessageClientIDSetter.getUniqID(msg), e);
} catch (ThreadDeath e) {
log.error("Send message async ThreadDeath", e);
throw this.checkProducerException(msg.getTopic(), MessageClientIDSetter.getUniqID(msg), e);

Check warning on line 164 in eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java

View check run for this annotation

Codecov / codecov/patch

eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/producer/ProducerImpl.java#L158-L164 

Added lines #L158 - L164 were not covered by tests
} catch (Exception e) {
log.error(String.format("Send message async Exception, %s", msg), e);
throw this.checkProducerException(msg.getTopic(), MessageClientIDSetter.getUniqID(msg), e);
Expand Down
Loading