diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy index 41bcc70fe532..51b3cbbf178f 100644 --- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy +++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy @@ -734,7 +734,7 @@ class BeamModulePlugin implements Plugin { google_api_services_bigquery : "com.google.apis:google-api-services-bigquery:v2-rev20230812-$google_clients_version", // Keep version consistent with the version in google_cloud_resourcemanager, managed by google_cloud_platform_libraries_bom google_api_services_cloudresourcemanager : "com.google.apis:google-api-services-cloudresourcemanager:v1-rev20230806-$google_clients_version", - google_api_services_dataflow : "com.google.apis:google-api-services-dataflow:v1b3-rev20231203-$google_clients_version", + google_api_services_dataflow : "com.google.apis:google-api-services-dataflow:v1b3-rev20240113-$google_clients_version", google_api_services_healthcare : "com.google.apis:google-api-services-healthcare:v1-rev20240110-$google_clients_version", google_api_services_pubsub : "com.google.apis:google-api-services-pubsub:v1-rev20220904-$google_clients_version", // Keep version consistent with the version in google_cloud_nio, managed by google_cloud_platform_libraries_bom diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java index a95e78288819..1ad0ed08c081 100644 --- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java +++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java @@ -1782,8 +1782,10 @@ private void sendWorkerMessage() throws IOException { new StreamingScalingReport() .setActiveThreadCount(workUnitExecutor.activeCount()) .setActiveBundleCount(workUnitExecutor.elementsOutstanding()) + .setOutstandingBytes(workUnitExecutor.bytesOutstanding()) .setMaximumThreadCount(chooseMaximumNumberOfThreads()) - .setMaximumBundleCount(workUnitExecutor.maximumElementsOutstanding()); + .setMaximumBundleCount(workUnitExecutor.maximumElementsOutstanding()) + .setMaximumBytes(workUnitExecutor.maximumBytesOutstanding()); workUnitClient.reportWorkerMessage( workUnitClient.createWorkerMessageFromStreamingScalingReport(activeThreadsReport)); } diff --git a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/DataflowWorkUnitClientTest.java b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/DataflowWorkUnitClientTest.java index 7720de3563b8..7019217b9fbd 100644 --- a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/DataflowWorkUnitClientTest.java +++ b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/DataflowWorkUnitClientTest.java @@ -240,7 +240,13 @@ public void testReportWorkerMessage() throws Exception { response.setContent(workerMessage.toPrettyString()); when(request.execute()).thenReturn(response); StreamingScalingReport activeThreadsReport = - new StreamingScalingReport().setActiveThreadCount(1); + new StreamingScalingReport() + .setActiveThreadCount(1) + .setActiveBundleCount(2) + .setOutstandingBytes(3L) + .setMaximumThreadCount(4) + .setMaximumBundleCount(5) + .setMaximumBytes(6L); WorkUnitClient client = new DataflowWorkUnitClient(pipelineOptions, LOG); WorkerMessage msg = client.createWorkerMessageFromStreamingScalingReport(activeThreadsReport); client.reportWorkerMessage(msg);