Re: [PR] add shutdown and start mechanics to windmill streams [beam]

[via GitHub]

m-trieu commented on code in PR #32774:
URL: https://github.com/apache/beam/pull/32774#discussion_r1810933751


##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -204,17 +224,23 @@ private void flushInternal(Map<Long, PendingRequest> 
requests) {
     }
   }
 
-  private void issueSingleRequest(final long id, PendingRequest 
pendingRequest) {
+  private void issueSingleRequest(long id, PendingRequest pendingRequest) {
     StreamingCommitWorkRequest.Builder requestBuilder = 
StreamingCommitWorkRequest.newBuilder();
     requestBuilder
         .addCommitChunkBuilder()
-        .setComputationId(pendingRequest.computation)
+        .setComputationId(pendingRequest.computationId())
         .setRequestId(id)
-        .setShardingKey(pendingRequest.request.getShardingKey())
-        .setSerializedWorkItemCommit(pendingRequest.request.toByteString());
+        .setShardingKey(pendingRequest.shardingKey())
+        .setSerializedWorkItemCommit(pendingRequest.serializedCommit());
     StreamingCommitWorkRequest chunk = requestBuilder.build();
     synchronized (this) {
-      pending.put(id, pendingRequest);
+      synchronized (shutdownLock) {
+        if (!isShutdown()) {
+          pending.put(id, pendingRequest);

Review Comment:
   done



##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -229,17 +255,24 @@ private void issueBatchedRequest(Map<Long, 
PendingRequest> requests) {
     for (Map.Entry<Long, PendingRequest> entry : requests.entrySet()) {
       PendingRequest request = entry.getValue();
       StreamingCommitRequestChunk.Builder chunkBuilder = 
requestBuilder.addCommitChunkBuilder();
-      if (lastComputation == null || 
!lastComputation.equals(request.computation)) {
-        chunkBuilder.setComputationId(request.computation);
-        lastComputation = request.computation;
+      if (lastComputation == null || 
!lastComputation.equals(request.computationId())) {
+        chunkBuilder.setComputationId(request.computationId());
+        lastComputation = request.computationId();
       }
-      chunkBuilder.setRequestId(entry.getKey());
-      chunkBuilder.setShardingKey(request.request.getShardingKey());
-      chunkBuilder.setSerializedWorkItemCommit(request.request.toByteString());
+      chunkBuilder
+          .setRequestId(entry.getKey())
+          .setShardingKey(request.shardingKey())
+          .setSerializedWorkItemCommit(request.serializedCommit());
     }
     StreamingCommitWorkRequest request = requestBuilder.build();
     synchronized (this) {
-      pending.putAll(requests);
+      synchronized (shutdownLock) {
+        if (!isShutdown()) {
+          pending.putAll(requests);
+        } else {
+          return;

Review Comment:
   done



##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -249,11 +282,16 @@ private void issueBatchedRequest(Map<Long, 
PendingRequest> requests) {
   }
 
   private void issueMultiChunkRequest(final long id, PendingRequest 
pendingRequest) {
-    checkNotNull(pendingRequest.computation);
-    final ByteString serializedCommit = pendingRequest.request.toByteString();
-
+    checkNotNull(pendingRequest.computationId());
+    final ByteString serializedCommit = pendingRequest.serializedCommit();
     synchronized (this) {
-      pending.put(id, pendingRequest);
+      synchronized (shutdownLock) {
+        if (!isShutdown()) {
+          pending.put(id, pendingRequest);
+        } else {
+          return;

Review Comment:
   done



-- 
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-unsubscr...@beam.apache.org

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