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 109c49377 [ISSUE #4450] Add retry strategy for sourceWorker(#4451)
109c49377 is described below
commit 109c493776927e2a50549b7162f95ec8cdca4031
Author: Fabian Bao <[email protected]>
AuthorDate: Tue Sep 19 23:11:49 2023 +0800
[ISSUE #4450] Add retry strategy for sourceWorker(#4451)
* feat: Add retry strategy for sourceWorker.
* feat: Add retry strategy for sourceWorker.
* feat: Add retry strategy for sourceWorker.
---
.../apache/eventmesh/openconnect/SourceWorker.java | 33 ++++++++++++++--------
1 file changed, 22 insertions(+), 11 deletions(-)
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
index a1fbc08b1..28910ee63 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
@@ -65,6 +65,8 @@ public class SourceWorker implements ConnectorWorker {
private final Source source;
private final SourceConfig config;
+ private static final int MAX_RETRY_TIMES = 3;
+
private OffsetStorageWriterImpl offsetStorageWriter;
private OffsetStorageReaderImpl offsetStorageReader;
@@ -177,16 +179,27 @@ public class SourceWorker implements ConnectorWorker {
// todo: convert connectRecord to cloudevent
CloudEvent event = convertRecordToEvent(connectRecord);
Optional<RecordOffsetManagement.SubmittedPosition>
submittedRecordPosition = prepareToUpdateRecordOffset(connectRecord);
- Package sendResult = eventMeshTCPClient.publish(event, 3000);
- if (sendResult.getHeader().getCode() == 0) {
- // publish success
- // commit record
- this.source.commit(connectRecord);
-
submittedRecordPosition.ifPresent(RecordOffsetManagement.SubmittedPosition::ack);
- } else {
- // todo: retry or other strategy
- log.error("{} failed to send record to {}, failed record {}",
this, event.getSubject(), connectRecord);
+ int retryTimes = 0;
+ // retry until MAX_RETRY_TIMES is reached
+ while (retryTimes < MAX_RETRY_TIMES) {
+ try {
+ Package sendResult = eventMeshTCPClient.publish(event,
3000);
+ if (sendResult.getHeader().getCode() == 0) {
+ // publish success
+ // commit record
+ this.source.commit(connectRecord);
+
submittedRecordPosition.ifPresent(RecordOffsetManagement.SubmittedPosition::ack);
+ break;
+ }
+ retryTimes++;
+ log.warn("{} failed to send record to {}, retry times =
{}, failed record {}",
+ this, event.getSubject(), retryTimes,
connectRecord);
+ } catch (Throwable t) {
+ retryTimes++;
+ log.error("{} failed to send record to {}, retry times =
{}, failed record {}, throw {}",
+ this, event.getSubject(), retryTimes,
connectRecord, t.getMessage());
+ }
}
offsetManagement.awaitAllMessages(5000, TimeUnit.MILLISECONDS);
@@ -326,6 +339,4 @@ public class SourceWorker implements ConnectorWorker {
}
return true;
}
-
-
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
