becketqin commented on a change in pull request #9388: [FLINK-13230] [pubsub]
Add retry mechanism to acknowledging pubsub me…
URL: https://github.com/apache/flink/pull/9388#discussion_r334816810
##########
File path:
flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/BlockingGrpcPubSubSubscriber.java
##########
@@ -95,12 +95,24 @@ public void acknowledge(List<String> acknowledgementIds) {
.addAllAckIds(splittedAckIds.f0)
.build();
- stub.withDeadlineAfter(60,
SECONDS).acknowledge(acknowledgeRequest);
-
+ acknowledgeWithRetries(acknowledgeRequest, retries);
splittedAckIds = splitAckIds(splittedAckIds.f1);
}
}
+ private void acknowledgeWithRetries(AcknowledgeRequest
acknowledgeRequest, int retriesRemaining) {
+ try {
+ stub.withDeadlineAfter(timeout.toMillis(),
TimeUnit.MILLISECONDS).acknowledge(acknowledgeRequest);
+ } catch (StatusRuntimeException e) {
+ if (retriesRemaining > 0) {
+ acknowledgeWithRetries(acknowledgeRequest,
retriesRemaining - 1);
Review comment:
We probably don't want to have recursion in the retry logic. Can we replace
it with a for loop instead?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
