Re: [PR] Worker message plumbing [beam]

["scwhittle (via GitHub)" <gi...@apache.org>]

scwhittle commented on code in PR #29879:
URL: https://github.com/apache/beam/pull/29879#discussion_r1440596051


##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java:
##########
@@ -1747,6 +1748,14 @@ private void updateThreadMetrics() {
     maxOutstandingBundles.addValue(workUnitExecutor.maximumElementsOutstanding());
   }
 
+  private void sendWorkerMessage() throws IOException {
+    LOG.info("[chengedward] creating StreamingScalingReport");

Review Comment:
   make sure to remove debug lgos



##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java:
##########
@@ -1758,6 +1767,13 @@ public void reportPeriodicWorkerUpdates() {
     } catch (Exception e) {
       LOG.error("Unexpected exception while trying to send counter updates", e);
     }
+    try {
+      sendWorkerMessage();

Review Comment:
   maybe do this from a separate periodic method than the existing method sending the counters?
   So that the latency of reporting this message doesn't hold up counters which can affect watermark latency in appliance.



##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowWorkUnitClient.java:
##########
@@ -269,4 +277,42 @@ public WorkItemServiceState reportWorkItemStatus(WorkItemStatus workItemStatus)
     logger.debug("ReportWorkItemStatus result: {}", state);
     return state;
   }
+
+  /** Reports the autoscaling signals to dataflow */
+  @Override
+  public WorkerMessageResponse reportWorkerMessage(StreamingScalingReport report)
+      throws IOException {
+    DateTime endTime = DateTime.now();
+    logger.debug("Reporting WorkMessageResponse");
+    Map<String, String> labels =
+        new HashMap<String, String>(
+            ImmutableMap.of("JOB_ID", options.getJobId(), "WORKER_ID", options.getWorkerId()));
+    WorkerMessage msg =
+        new WorkerMessage()
+            .setTime(toCloudTime(endTime))
+            .setStreamingScalingReport(report)
+            .setLabels(labels);
+    SendWorkerMessagesRequest request =
+        new SendWorkerMessagesRequest()
+            .setLocation(options.getRegion())
+            .setWorkerMessages(Collections.singletonList(msg));
+    SendWorkerMessagesResponse result =
+        dataflow
+            .projects()
+            .locations()
+            .workerMessages(options.getProject(), options.getRegion(), request)
+            .execute();
+    if (result == null) {
+      logger.warn("Worker Message response is null");
+      throw new IOException("Got null Worker Message response");
+    }
+
+    // Currently no response is expected
+    if (result.getWorkerMessageResponses() == null) {
+      return new WorkerMessageResponse();
+    }
+    WorkerMessageResponse response = result.getWorkerMessageResponses().get(0);

Review Comment:
   verify there is either 1 or 0 responses? this will just ignore more than the first which could be confusing.



##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowWorkUnitClient.java:
##########
@@ -269,4 +277,42 @@ public WorkItemServiceState reportWorkItemStatus(WorkItemStatus workItemStatus)
     logger.debug("ReportWorkItemStatus result: {}", state);
     return state;
   }
+
+  /** Reports the autoscaling signals to dataflow */
+  @Override
+  public WorkerMessageResponse reportWorkerMessage(StreamingScalingReport report)
+      throws IOException {
+    DateTime endTime = DateTime.now();
+    logger.debug("Reporting WorkMessageResponse");
+    Map<String, String> labels =
+        new HashMap<String, String>(

Review Comment:
   rm new HashMap , that copies entries from the immutable map but then will be mutable.  you can just use the immutable map



##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java:
##########
@@ -1758,6 +1767,13 @@ public void reportPeriodicWorkerUpdates() {
     } catch (Exception e) {
       LOG.error("Unexpected exception while trying to send counter updates", e);
     }
+    try {
+      sendWorkerMessage();

Review Comment:
   do we want to do this for appliance or just streaming engine?



##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowWorkUnitClient.java:
##########
@@ -269,4 +277,42 @@ public WorkItemServiceState reportWorkItemStatus(WorkItemStatus workItemStatus)
     logger.debug("ReportWorkItemStatus result: {}", state);
     return state;
   }
+
+  /** Reports the autoscaling signals to dataflow */
+  @Override
+  public WorkerMessageResponse reportWorkerMessage(StreamingScalingReport report)

Review Comment:
   if this stays as is, the method should be renamed.  But I think it would be better if we can combine with other data sent in the same rpc.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org