Avoid creating Duration objects when direclty converting to millis (#36969)

Author: arunpandianp

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContext.java

@@ -30,6 +30,7 @@
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 import java.util.logging.LogRecord;
 import java.util.stream.Collectors;
@@ -365,7 +366,7 @@ private String getBundleLullMessage(Thread trackedThread, Duration lullDuration)
           message.append(
               "Time spent in this step(millis): "
                   + (clock.currentTimeMillis()
-                      - getActiveMessageMetadata().get().stopwatch().elapsed().toMillis())
+                      - getActiveMessageMetadata().get().stopwatch().elapsed(TimeUnit.MILLISECONDS))
                   + "\n");
         }
         message.append("Processing times in each step(millis)\n");
@@ -476,7 +477,8 @@ private synchronized void recordActiveMessageInProcessingTimesMap() {
       if (this.activeMessageMetadata == null) {
         return;
       }
-      int processingTime = (int) (this.activeMessageMetadata.stopwatch().elapsed().toMillis());
+      int processingTime =
+          (int) (this.activeMessageMetadata.stopwatch().elapsed(TimeUnit.MILLISECONDS));
       this.processingTimesByStep.compute(
           this.activeMessageMetadata.userStepName(),
           (k, v) -> {

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/Work.java

@@ -26,6 +26,7 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 import javax.annotation.concurrent.NotThreadSafe;
@@ -151,7 +152,7 @@ private static LatencyAttribution.Builder createLatencyAttributionWithActiveLate
       stepBuilder.setUserStepName(activeMessage.get().userStepName());
       ActiveElementMetadata.Builder activeElementBuilder = ActiveElementMetadata.newBuilder();
       activeElementBuilder.setProcessingTimeMillis(
-          activeMessage.get().stopwatch().elapsed().toMillis());
+          activeMessage.get().stopwatch().elapsed(TimeUnit.MILLISECONDS));
       stepBuilder.setActiveMessageMetadata(activeElementBuilder);
       latencyAttribution.addActiveLatencyBreakdown(stepBuilder.build());
       return latencyAttribution;