Skip to content

Commit 6c98c3d

Browse files
committed
address comments
1 parent 4b5f166 commit 6c98c3d

1 file changed

Lines changed: 4 additions & 4 deletions

File tree

  • runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -243,23 +243,23 @@ public boolean hasPendingRequests() {
243243
public void onDone(Status status) {
244244
if (maxRetryDuration.compareTo(Duration.ZERO) > 0) {
245245
// Remove the requests that have exceeded the retry time so they are not retried.
246-
long startTimeRetryThresholdNanos = System.nanoTime() - maxRetryDuration.toNanos();
246+
long nowNanos = System.nanoTime();
247+
long maxRetryDurationNanos = maxRetryDuration.toNanos();
247248
Iterator<Map.Entry<Long, StreamAndRequest>> iterator = pending.entrySet().iterator();
248249
int keptRequests = 0, removedRequests = 0;
249250
while (iterator.hasNext()) {
250251
StreamAndRequest streamAndRequest = checkNotNull(iterator.next().getValue());
251252
PendingRequest pendingRequest = streamAndRequest.request;
252253
if (!belongsToThisHandler(streamAndRequest)
253-
|| pendingRequest.getStartTimeNanos() > startTimeRetryThresholdNanos) {
254+
|| nowNanos - pendingRequest.getStartTimeNanos() < maxRetryDurationNanos) {
254255
++keptRequests;
255256
continue;
256257
}
257258
++removedRequests;
259+
iterator.remove();
258260
try {
259261
pendingRequest.completeWithStatus(CommitStatus.ABORTED);
260-
iterator.remove();
261262
} catch (RuntimeException e) {
262-
// Ignore exceptions and retry the commit.
263263
LOG.warn("Exception while aborting commit due to retry timeout.", e);
264264
}
265265
}

0 commit comments

Comments
 (0)