address comments

Author: scwhittle

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowStreamingPipelineOptions.java

@@ -201,7 +201,7 @@ public interface DataflowStreamingPipelineOptions extends PipelineOptions {
   @Default.InstanceFactory(CommitWorkStreamRetryTimeoutMillisFactory.class)
   long getCommitWorkStreamRetryTimeoutMillis();
 
-  void getCommitWorkStreamRetryTimeoutMillis(long value);
+  void setCommitWorkStreamRetryTimeoutMillis(long value);
 
   @Description(
       "Period for sending 'global get config' requests to the service. The duration is "
@@ -351,7 +351,6 @@ public Boolean create(PipelineOptions options) {
     }
   }
 
-  /** defaults to false unless one of the experiment is set. */
   class CommitWorkStreamRetryTimeoutMillisFactory implements DefaultValueFactory<Long> {
     @Override
     public Long create(PipelineOptions options) {

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

@@ -243,27 +243,25 @@ public boolean hasPendingRequests() {
     public void onDone(Status status) {
       if (maxRetryDuration.compareTo(Duration.ZERO) > 0) {
         // Remove the requests that have exceeded the retry time so they are not retried.
-        long startTimeRetryThresholdMsec = System.currentTimeMillis() - maxRetryDuration.toMillis();
+        long startTimeRetryThresholdNanos = System.nanoTime() - maxRetryDuration.toNanos();
         Iterator<Map.Entry<Long, StreamAndRequest>> iterator = pending.entrySet().iterator();
         int keptRequests = 0, removedRequests = 0;
         while (iterator.hasNext()) {
           StreamAndRequest streamAndRequest = checkNotNull(iterator.next().getValue());
           PendingRequest pendingRequest = streamAndRequest.request;
           if (!belongsToThisHandler(streamAndRequest)
-              || pendingRequest.getStartTimeMillis() > startTimeRetryThresholdMsec) {
+              || pendingRequest.getStartTimeNanos() > startTimeRetryThresholdNanos) {
             ++keptRequests;
             continue;
           }
           ++removedRequests;
           try {
             pendingRequest.completeWithStatus(CommitStatus.ABORTED);
+            iterator.remove();
           } catch (RuntimeException e) {
-            // Catch possible exceptions to ensure that an exception for one commit does not prevent
-            // other commits from being processed. Aggregate all the failures to throw after
-            // processing the response if they exist.
+            // Ignore exceptions and retry the commit.
             LOG.warn("Exception while aborting commit due to retry timeout.", e);
           }
-          iterator.remove();
         }
         if (removedRequests > 0) {
           LOG.info(
@@ -420,14 +418,14 @@ private static class PendingRequest {
     private final String computationId;
     private final WorkItemCommitRequest request;
     private final Consumer<CommitStatus> onDone;
-    private final long startTimeMillis;
+    private final long startTimeNanos; // System.nanoTime() of when request began.
 
     private PendingRequest(
         String computationId, WorkItemCommitRequest request, Consumer<CommitStatus> onDone) {
       this.computationId = computationId;
       this.request = request;
       this.onDone = onDone;
-      this.startTimeMillis = System.currentTimeMillis();
+      this.startTimeNanos = System.nanoTime();
     }
 
     String getComputationId() {
@@ -438,8 +436,8 @@ WorkItemCommitRequest getRequest() {
       return request;
     }
 
-    long getStartTimeMillis() {
-      return startTimeMillis;
+    long getStartTimeNanos() {
+      return startTimeNanos;
     }
 
     private long getBytes() {