Merge pull request #33225 Take maxBufferingDuration payload from mill…

…is directly Changes maxBufferingDuration payload to be taken from millis directly…
apache · Dec 4, 2024 · ed163f6 · ed163f6
2 parents a9a82a1 + 54eb50d
commit ed163f6
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 3 deletions.
diff --git a/...core/src/main/java/org/apache/beam/sdk/util/construction/GroupIntoBatchesTranslation.java b/...core/src/main/java/org/apache/beam/sdk/util/construction/GroupIntoBatchesTranslation.java
@@ -83,7 +83,7 @@ private static <K, V> GroupIntoBatchesPayload getPayloadFromParameters(
     return RunnerApi.GroupIntoBatchesPayload.newBuilder()
         .setBatchSize(params.getBatchSize())
         .setBatchSizeBytes(params.getBatchSizeBytes())
-        .setMaxBufferingDurationMillis(params.getMaxBufferingDuration().getStandardSeconds() * 1000)
+        .setMaxBufferingDurationMillis(params.getMaxBufferingDuration().getMillis())
         .build();
   }
 

diff --git a/.../src/test/java/org/apache/beam/sdk/util/construction/GroupIntoBatchesTranslationTest.java b/.../src/test/java/org/apache/beam/sdk/util/construction/GroupIntoBatchesTranslationTest.java
@@ -71,6 +71,7 @@ public static Iterable<GroupIntoBatches<String, Integer>> transform() {
             ImmutableSet.of( //
                 gib -> gib, //
                 gib -> gib.withMaxBufferingDuration(Duration.ZERO), //
+                gib -> gib.withMaxBufferingDuration(Duration.millis(200)), //
                 gib -> gib.withMaxBufferingDuration(Duration.standardSeconds(10)));
 
     return Sets.cartesianProduct(
@@ -150,7 +151,7 @@ private void verifyPayload(
     assertThat(payload.getBatchSize(), equalTo(params.getBatchSize()));
     assertThat(payload.getBatchSizeBytes(), equalTo(params.getBatchSizeBytes()));
     assertThat(
-        payload.getMaxBufferingDurationMillis(),
-        equalTo(params.getMaxBufferingDuration().getStandardSeconds() * 1000));
+        Duration.millis(payload.getMaxBufferingDurationMillis()),
+        equalTo(params.getMaxBufferingDuration()));
   }
 }