You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "arunpandianp (via GitHub)" <gi...@apache.org> on 2026/06/02 11:38:34 UTC
[PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
arunpandianp opened a new pull request, #38768:
URL: https://github.com/apache/beam/pull/38768
- Add MultiKeyWorkItemCommitRequest to windmill.proto.
- Support MultiKey commits in Commit model and StreamingEngineWorkCommitter.
- Update GrpcCommitWorkStream to batch and stream MultiKey commit requests.
--
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
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #38768:
URL: https://github.com/apache/beam/pull/38768#discussion_r3636814668
##########
runners/google-cloud-dataflow-java/worker/windmill/src/main/proto/windmill.proto:
##########
@@ -678,9 +678,24 @@ message WorkItemCommitRequest {
reserved 6, 23;
}
+message MultiKeyWorkItemCommitRequest {
+ optional Uint128Proto key_group = 7;
+
+ repeated WorkItemCommitRequest requests = 1;
+
+ repeated OutputMessageBundle output_messages = 2;
+
+ repeated PubSubMessageBundle pubsub_messages = 3;
+
+ repeated int64 finalize_ids = 4 [packed = true];
+
+ reserved 6;
Review Comment:
This is matching the server side protos. 5 is also missing there, think it was skipped accidentally.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/Commit.java:
##########
@@ -32,20 +35,43 @@ public abstract class Commit {
public static Commit create(
WorkItemCommitRequest request, ComputationState computationState, Work work) {
Preconditions.checkArgument(request.getSerializedSize() > 0);
- return new AutoValue_Commit(request, computationState, work);
+ return new AutoValue_Commit(
+ Optional.of(request), computationState, Optional.empty(), ImmutableList.of(work));
+ }
+
+ public static Commit createMultiKey(
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest,
+ ComputationState computationState,
+ ImmutableList<Work> workBatch) {
+ Preconditions.checkArgument(!workBatch.isEmpty());
+ return new AutoValue_Commit(
+ Optional.empty(), computationState, Optional.of(multiKeyRequest), workBatch);
}
public final String computationId() {
return computationState().getComputationId();
}
- public abstract WorkItemCommitRequest request();
+ public abstract Optional<WorkItemCommitRequest> singleKeyRequest();
public abstract ComputationState computationState();
- public abstract Work work();
+ public abstract Optional<Windmill.MultiKeyWorkItemCommitRequest> multiKeyRequest();
+
+ public abstract ImmutableList<Work> workBatch();
+
+ public final boolean isFailed() {
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #38768:
URL: https://github.com/apache/beam/pull/38768#discussion_r3625878499
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitter.java:
##########
@@ -113,8 +114,8 @@ public void commit(Commit commit) {
"Trying to queue commit on shutdown, failing commit=[computationId={}, shardingKey={},"
+ " workId={} ].",
commit.computationId(),
- commit.work().getShardedKey(),
- commit.work().id());
+ commit.workBatch().get(0).getShardedKey(),
Review Comment:
I was going to say to check the size before blind access here, but maybe we should just add a toString for the Commit instead and use it here?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/Commit.java:
##########
@@ -32,20 +35,43 @@ public abstract class Commit {
public static Commit create(
WorkItemCommitRequest request, ComputationState computationState, Work work) {
Preconditions.checkArgument(request.getSerializedSize() > 0);
- return new AutoValue_Commit(request, computationState, work);
+ return new AutoValue_Commit(
+ Optional.of(request), computationState, Optional.empty(), ImmutableList.of(work));
+ }
+
+ public static Commit createMultiKey(
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest,
+ ComputationState computationState,
+ ImmutableList<Work> workBatch) {
+ Preconditions.checkArgument(!workBatch.isEmpty());
+ return new AutoValue_Commit(
+ Optional.empty(), computationState, Optional.of(multiKeyRequest), workBatch);
}
public final String computationId() {
return computationState().getComputationId();
}
- public abstract WorkItemCommitRequest request();
+ public abstract Optional<WorkItemCommitRequest> singleKeyRequest();
public abstract ComputationState computationState();
- public abstract Work work();
+ public abstract Optional<Windmill.MultiKeyWorkItemCommitRequest> multiKeyRequest();
+
+ public abstract ImmutableList<Work> workBatch();
+
+ public final boolean isFailed() {
+ for (Work w : workBatch()) {
+ if (w.isFailed()) {
+ return true;
+ }
+ }
+ return false;
+ }
public final int getSize() {
Review Comment:
since we now maybe have multiple, size is confusing. How about serializedByteSize or something more explicit
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/Commit.java:
##########
@@ -32,20 +35,43 @@ public abstract class Commit {
public static Commit create(
WorkItemCommitRequest request, ComputationState computationState, Work work) {
Preconditions.checkArgument(request.getSerializedSize() > 0);
- return new AutoValue_Commit(request, computationState, work);
+ return new AutoValue_Commit(
+ Optional.of(request), computationState, Optional.empty(), ImmutableList.of(work));
+ }
+
+ public static Commit createMultiKey(
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest,
+ ComputationState computationState,
+ ImmutableList<Work> workBatch) {
+ Preconditions.checkArgument(!workBatch.isEmpty());
+ return new AutoValue_Commit(
+ Optional.empty(), computationState, Optional.of(multiKeyRequest), workBatch);
}
public final String computationId() {
return computationState().getComputationId();
}
- public abstract WorkItemCommitRequest request();
+ public abstract Optional<WorkItemCommitRequest> singleKeyRequest();
Review Comment:
should we remove AutoValue and internally use a boolean and nullable references instead of Optional?
We can avoid the Optional allocations in hot path and can also enforce that only one or the other is set.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitter.java:
##########
@@ -147,8 +148,12 @@ private void drainCommitQueue() {
}
private void failCommit(Commit commit) {
Review Comment:
should we name this failQueuedCommit? Might be clearer that we don't need to update the state or the active bytes
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -512,7 +528,34 @@ public boolean commitWorkItem(
return false;
}
- PendingRequest request = new PendingRequest(computation, commitRequest, onDone);
+ PendingRequest request =
+ new PendingRequest(
+ computation,
+ commitRequest.getShardingKey(),
+ commitRequest.toByteString(),
+ StreamingCommitRequestChunk.CommitType.COMMIT_TYPE_SINGLE_KEY,
+ onDone);
+ add(idGenerator.incrementAndGet(), request);
+ return true;
+ }
+
+ @Override
+ public boolean commitMultiKeyWorkItem(
+ String computation,
+ Windmill.MultiKeyWorkItemCommitRequest commitRequest,
+ Consumer<CommitStatus> onDone) {
+ if (!canAccept(commitRequest.getSerializedSize() + computation.length())) {
+ return false;
+ }
+ Preconditions.checkArgument(commitRequest.getRequestsCount() > 0);
Review Comment:
put top of function?
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingApplianceWorkCommitterTest.java:
##########
@@ -141,12 +141,13 @@ public void testCommit() {
(CompleteCommit completeCommit, Commit commit) ->
completeCommit.computationId().equals(commit.computationId())
&& completeCommit.status() == Windmill.CommitStatus.OK
- && completeCommit.workId().equals(commit.work().id())
+ && completeCommit.workId().equals(commit.workBatch().get(0).id())
Review Comment:
add && commitWorkBatch.size() == 1
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitterTest.java:
##########
@@ -409,10 +436,14 @@ public void testMultipleCommitSendersSingleStream() {
waitForExpectedSetSize(completeCommits, commits.size());
for (Commit commit : commits) {
- WorkItemCommitRequest request = committed.get(commit.work().getWorkItem().getWorkToken());
+ WorkItemCommitRequest request =
Review Comment:
verify batch size
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitterTest.java:
##########
@@ -474,4 +505,201 @@ public void testStop_drainsCommitQueue_concurrentCommit()
waitForExpectedSetSize(completeCommits, sentCommits.intValue());
}
+
+ @Test
+ public void testCommit_multiKeyCommitFailedWork() {
+ Set<CompleteCommit> completeCommits = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ workCommitter = createWorkCommitter(completeCommits::add);
+
+ Work workA = createMockWork(101L);
+ Work workB = createMockWork(102L);
+ Work workC = createMockWork(103L);
+
+ // Mark non-primary key B as failed
+ workB.setFailed();
+
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest =
+ Windmill.MultiKeyWorkItemCommitRequest.newBuilder()
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workA.getWorkItem().getKey())
+ .setShardingKey(workA.getWorkItem().getShardingKey())
+ .setWorkToken(workA.getWorkItem().getWorkToken())
+ .setCacheToken(workA.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workB.getWorkItem().getKey())
+ .setShardingKey(workB.getWorkItem().getShardingKey())
+ .setWorkToken(workB.getWorkItem().getWorkToken())
+ .setCacheToken(workB.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workC.getWorkItem().getKey())
+ .setShardingKey(workC.getWorkItem().getShardingKey())
+ .setWorkToken(workC.getWorkItem().getWorkToken())
+ .setCacheToken(workC.getWorkItem().getCacheToken())
+ .build())
+ .build();
+
+ Commit commit =
+ Commit.createMultiKey(
+ multiKeyRequest,
+ createComputationState("computationId"),
+ ImmutableList.of(workA, workB, workC));
+
+ workCommitter.start();
+ workCommitter.commit(commit);
+
+ // The entire batch must be aborted immediately without making network calls
+ waitForExpectedSetSize(completeCommits, 3);
+
+ // Verify all three works are aborted individually
+ assertThat(completeCommits)
+ .containsExactly(
+ CompleteCommit.create(
+ "computationId", workA.getShardedKey(), workA.id(), CommitStatus.ABORTED),
+ CompleteCommit.create(
+ "computationId", workB.getShardedKey(), workB.id(), CommitStatus.ABORTED),
+ CompleteCommit.create(
+ "computationId", workC.getShardedKey(), workC.id(), CommitStatus.ABORTED));
+
+ workCommitter.stop();
+ }
+
+ @Test
+ public void testCommit_multiKeyCommitSuccess() {
+ Set<CompleteCommit> completeCommits = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ workCommitter = createWorkCommitter(completeCommits::add);
+
+ Work workA = createMockWork(101L);
+ Work workB = createMockWork(102L);
+ Work workC = createMockWork(103L);
+
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest =
+ Windmill.MultiKeyWorkItemCommitRequest.newBuilder()
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workA.getWorkItem().getKey())
+ .setShardingKey(workA.getWorkItem().getShardingKey())
+ .setWorkToken(workA.getWorkItem().getWorkToken())
+ .setCacheToken(workA.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workB.getWorkItem().getKey())
+ .setShardingKey(workB.getWorkItem().getShardingKey())
+ .setWorkToken(workB.getWorkItem().getWorkToken())
+ .setCacheToken(workB.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workC.getWorkItem().getKey())
+ .setShardingKey(workC.getWorkItem().getShardingKey())
+ .setWorkToken(workC.getWorkItem().getWorkToken())
+ .setCacheToken(workC.getWorkItem().getCacheToken())
+ .build())
+ .build();
+
+ Commit commit =
+ Commit.createMultiKey(
+ multiKeyRequest,
+ createComputationState("computationId"),
+ ImmutableList.of(workA, workB, workC));
+
+ workCommitter.start();
+ workCommitter.commit(commit);
+
+ // Wait for the server to receive and process the commits
+ fakeWindmillServer.waitForAndGetCommits(3);
+ waitForExpectedSetSize(completeCommits, 3);
+
+ // Verify that FakeWindmillServer received all 3 work requests in multiKeyCommitsReceived
+ List<Windmill.MultiKeyWorkItemCommitRequest> multiKeyCommits =
+ fakeWindmillServer.getMultiKeyCommitsReceived();
+ assertThat(multiKeyCommits).hasSize(1);
+ assertThat(multiKeyCommits.get(0)).isEqualTo(multiKeyRequest);
+
+ // Verify all three works are completed successfully
+ assertThat(completeCommits)
+ .containsExactly(
+ CompleteCommit.create(
+ "computationId", workA.getShardedKey(), workA.id(), CommitStatus.OK),
+ CompleteCommit.create(
+ "computationId", workB.getShardedKey(), workB.id(), CommitStatus.OK),
+ CompleteCommit.create(
+ "computationId", workC.getShardedKey(), workC.id(), CommitStatus.OK));
+
+ workCommitter.stop();
+ }
+
+ @Test
+ public void testCommit_multiKeyCommitStatusNotOK() {
+ Set<CompleteCommit> completeCommits = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ workCommitter = createWorkCommitter(completeCommits::add);
+
+ Work workA = createMockWork(101L);
+ Work workB = createMockWork(102L);
+ Work workC = createMockWork(103L);
+
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest =
+ Windmill.MultiKeyWorkItemCommitRequest.newBuilder()
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workA.getWorkItem().getKey())
+ .setShardingKey(workA.getWorkItem().getShardingKey())
+ .setWorkToken(workA.getWorkItem().getWorkToken())
+ .setCacheToken(workA.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workB.getWorkItem().getKey())
+ .setShardingKey(workB.getWorkItem().getShardingKey())
+ .setWorkToken(workB.getWorkItem().getWorkToken())
+ .setCacheToken(workB.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workC.getWorkItem().getKey())
+ .setShardingKey(workC.getWorkItem().getShardingKey())
+ .setWorkToken(workC.getWorkItem().getWorkToken())
+ .setCacheToken(workC.getWorkItem().getCacheToken())
+ .build())
+ .build();
+
+ Commit commit =
+ Commit.createMultiKey(
+ multiKeyRequest,
+ createComputationState("computationId"),
+ ImmutableList.of(workA, workB, workC));
+
+ // Offer NOT_FOUND status for one of the works.
+ fakeWindmillServer.whenCommitWorkStreamCalled().put(workB.id(), CommitStatus.NOT_FOUND);
+
+ workCommitter.start();
+ workCommitter.commit(commit);
+
+ // Wait for the server to receive and process the commits
+ fakeWindmillServer.waitForAndGetCommits(3);
+ waitForExpectedSetSize(completeCommits, 3);
+
+ // Verify that FakeWindmillServer received the multi-key commit
+ List<Windmill.MultiKeyWorkItemCommitRequest> multiKeyCommits =
+ fakeWindmillServer.getMultiKeyCommitsReceived();
+ assertThat(multiKeyCommits).hasSize(1);
+ assertThat(multiKeyCommits.get(0)).isEqualTo(multiKeyRequest);
+
+ // Verify all three works in the multi-key commit are completed with NOT_FOUND status
Review Comment:
this seems more like it is testing the fakeWindmillServer than the StreamingEngineCommitter, it is the one fanning out from single multicommit to separate CompleteCommit.
If you want to test StreamingEngineCommitter it seems like it should verify that the error doesn't result in the committer retrying or something
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/FakeWindmillServer.java:
##########
@@ -445,6 +460,37 @@ public void flush() {
.orElse(Windmill.CommitStatus.OK));
}
requests.clear();
+
+ for (MultiKeyRequestAndDone elem : multiKeyRequests) {
+ if (dropStreamingCommits) {
+ for (WorkItemCommitRequest workRequest : elem.request.getRequestsList()) {
+ droppedStreamingCommits.put(workRequest.getWorkToken(), elem.onDone);
+ }
+ continue;
+ }
+
+ multiKeyCommitsReceived.add(elem.request);
+ for (WorkItemCommitRequest workRequest : elem.request.getRequestsList()) {
+ commitsReceived.put(workRequest.getWorkToken(), workRequest);
+ }
+
+ // Determine status for the batch.
+ // Default to OK, but if any of the works in the batch has an offered status, use it.
+ Windmill.CommitStatus status = Windmill.CommitStatus.OK;
+ for (WorkItemCommitRequest workRequest : elem.request.getRequestsList()) {
+ Windmill.CommitStatus offeredStatus =
+ streamingCommitsToOffer.remove(
Review Comment:
instead of this merging should we instead have a way to set the status for multi-key requests?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/Commit.java:
##########
@@ -32,20 +35,43 @@ public abstract class Commit {
public static Commit create(
WorkItemCommitRequest request, ComputationState computationState, Work work) {
Preconditions.checkArgument(request.getSerializedSize() > 0);
- return new AutoValue_Commit(request, computationState, work);
+ return new AutoValue_Commit(
+ Optional.of(request), computationState, Optional.empty(), ImmutableList.of(work));
+ }
+
+ public static Commit createMultiKey(
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest,
+ ComputationState computationState,
+ ImmutableList<Work> workBatch) {
+ Preconditions.checkArgument(!workBatch.isEmpty());
+ return new AutoValue_Commit(
+ Optional.empty(), computationState, Optional.of(multiKeyRequest), workBatch);
}
public final String computationId() {
return computationState().getComputationId();
}
- public abstract WorkItemCommitRequest request();
+ public abstract Optional<WorkItemCommitRequest> singleKeyRequest();
public abstract ComputationState computationState();
- public abstract Work work();
+ public abstract Optional<Windmill.MultiKeyWorkItemCommitRequest> multiKeyRequest();
+
+ public abstract ImmutableList<Work> workBatch();
+
+ public final boolean isFailed() {
Review Comment:
Maybe this should be moved to StreamingEngineWorkCommitter just based upon workbatch?
I think that would be clearer next to the failCommit logic which also works on the batch.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingApplianceWorkCommitterTest.java:
##########
@@ -129,9 +129,9 @@ public void testCommit() {
for (Commit commit : commits) {
Windmill.WorkItemCommitRequest request =
- committed.get(commit.work().getWorkItem().getWorkToken());
+ committed.get(commit.workBatch().get(0).getWorkItem().getWorkToken());
Review Comment:
assert workBatch size is 1 also
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitterTest.java:
##########
@@ -186,10 +186,14 @@ public void testCommit_sendsCommitsToStreamingEngine() {
waitForExpectedSetSize(completeCommits, 5);
for (Commit commit : commits) {
- WorkItemCommitRequest request = committed.get(commit.work().getWorkItem().getWorkToken());
+ WorkItemCommitRequest request =
+ committed.get(commit.workBatch().get(0).getWorkItem().getWorkToken());
Review Comment:
verify the batch size is 1
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitterTest.java:
##########
@@ -224,14 +228,24 @@ public void testCommit_handlesFailedCommits() {
waitForExpectedSetSize(completeCommits, 10);
for (Commit commit : commits) {
- if (commit.work().isFailed()) {
+ if (commit.isFailed()) {
Review Comment:
verify the batch size is 1
##########
runners/google-cloud-dataflow-java/worker/windmill/src/main/proto/windmill.proto:
##########
@@ -678,9 +678,24 @@ message WorkItemCommitRequest {
reserved 6, 23;
}
+message MultiKeyWorkItemCommitRequest {
+ optional Uint128Proto key_group = 7;
+
+ repeated WorkItemCommitRequest requests = 1;
+
+ repeated OutputMessageBundle output_messages = 2;
+
+ repeated PubSubMessageBundle pubsub_messages = 3;
+
+ repeated int64 finalize_ids = 4 [packed = true];
+
+ reserved 6;
Review Comment:
is the reserved needed?
what about 5?
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/FakeWindmillServer.java:
##########
@@ -423,6 +426,18 @@ public boolean commitWorkItem(
return true;
}
+ @Override
+ public boolean commitMultiKeyWorkItem(
+ String computation,
+ Windmill.MultiKeyWorkItemCommitRequest request,
+ Consumer<Windmill.CommitStatus> onDone) {
+ LOG.debug("commitWorkStream::commitMultiKeyWorkItem: {}", request);
+ if (multiKeyRequests.size() > 5) return false;
Review Comment:
is this used by any test? could we have a constant for this magic value if so?
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitterTest.java:
##########
@@ -282,11 +296,16 @@ public void testCommit_handlesCompleteCommits_commitStatusNotOK() {
waitForExpectedSetSize(completeCommits, commits.size());
for (Commit commit : commits) {
- WorkItemCommitRequest request = committed.get(commit.work().getWorkItem().getWorkToken());
+ WorkItemCommitRequest request =
Review Comment:
ditto
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStreamTest.java:
##########
@@ -1134,6 +1134,76 @@ public void testCommitWorkItem_multiplePhysicalStreams_multipleHandovers_halfClo
assertTrue(commitWorkStream.awaitTermination(10, TimeUnit.SECONDS));
}
+ @Test
+ public void testCommit_multiKeyCommit() throws Exception {
+ GrpcCommitWorkStream commitWorkStream = createCommitWorkStream();
+ FakeWindmillGrpcService.CommitStreamInfo streamInfo = waitForConnectionAndConsumeHeader();
+
+ CompletableFuture<Windmill.CommitStatus> commitStatusFuture = new CompletableFuture<>();
+
+ // 1. Construct two individual WorkItemCommitRequests
+ long shardingKey1 = 101L;
+ long workToken1 = 201L;
+ long cacheToken1 = 301L;
+ long shardingKey2 = 102L;
+ long workToken2 = 202L;
+ long cacheToken2 = 302L;
+ Windmill.WorkItemCommitRequest request1 =
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(ByteString.copyFromUtf8("key1"))
+ .setShardingKey(shardingKey1)
+ .setWorkToken(workToken1)
+ .setCacheToken(cacheToken1)
+ .build();
+ Windmill.WorkItemCommitRequest request2 =
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(ByteString.copyFromUtf8("key2"))
+ .setShardingKey(shardingKey2)
+ .setWorkToken(workToken2)
+ .setCacheToken(cacheToken2)
+ .build();
+
+ // 2. Wrap them into a MultiKeyWorkItemCommitRequest
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest =
+ Windmill.MultiKeyWorkItemCommitRequest.newBuilder()
+ .addRequests(request1)
+ .addRequests(request2)
+ .build();
+
+ // 3. Commit the multi-key work item using the request batcher
+ try (WindmillStream.CommitWorkStream.RequestBatcher batcher = commitWorkStream.batcher()) {
+ assertTrue(
+ batcher.commitMultiKeyWorkItem(
+ COMPUTATION_ID, multiKeyRequest, commitStatusFuture::complete));
+ }
+
+ // 4. Receive and assert request properties on FakeWindmillGrpcService
+ Windmill.StreamingCommitWorkRequest request = streamInfo.requests.take();
+ assertThat(request.getCommitChunkCount()).isEqualTo(1);
+
+ Windmill.StreamingCommitRequestChunk chunk = request.getCommitChunk(0);
+
+ // Assert that the commit type is correctly identified as COMMIT_TYPE_MULTI_KEY
+ assertThat(chunk.getCommitType())
+ .isEqualTo(Windmill.StreamingCommitRequestChunk.CommitType.COMMIT_TYPE_MULTI_KEY);
+
+ // Assert that the routing sharding key is mapped to the first request's sharding key
+ assertThat(chunk.getShardingKey()).isEqualTo(request1.getShardingKey());
+
+ // Assert that the serialized payload matches the input multiKeyRequest
+ Windmill.MultiKeyWorkItemCommitRequest parsedRequest =
+ Windmill.MultiKeyWorkItemCommitRequest.parseFrom(chunk.getSerializedWorkItemCommit());
+ assertThat(parsedRequest).isEqualTo(multiKeyRequest);
+
+ // 5. Respond with the generated requestId to complete the commit
+ long requestId = chunk.getRequestId();
+ streamInfo.responseObserver.onNext(
+ Windmill.StreamingCommitResponse.newBuilder().addRequestId(requestId).build());
+
+ // 6. Verify callback completed successfully with CommitStatus.OK
+ assertThat(commitStatusFuture.get()).isEqualTo(Windmill.CommitStatus.OK);
Review Comment:
similar test but return an error and verify plumbing?
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitterTest.java:
##########
@@ -474,4 +505,201 @@ public void testStop_drainsCommitQueue_concurrentCommit()
waitForExpectedSetSize(completeCommits, sentCommits.intValue());
}
+
+ @Test
+ public void testCommit_multiKeyCommitFailedWork() {
+ Set<CompleteCommit> completeCommits = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ workCommitter = createWorkCommitter(completeCommits::add);
+
+ Work workA = createMockWork(101L);
+ Work workB = createMockWork(102L);
+ Work workC = createMockWork(103L);
+
+ // Mark non-primary key B as failed
+ workB.setFailed();
+
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest =
+ Windmill.MultiKeyWorkItemCommitRequest.newBuilder()
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workA.getWorkItem().getKey())
+ .setShardingKey(workA.getWorkItem().getShardingKey())
+ .setWorkToken(workA.getWorkItem().getWorkToken())
+ .setCacheToken(workA.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workB.getWorkItem().getKey())
+ .setShardingKey(workB.getWorkItem().getShardingKey())
+ .setWorkToken(workB.getWorkItem().getWorkToken())
+ .setCacheToken(workB.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workC.getWorkItem().getKey())
+ .setShardingKey(workC.getWorkItem().getShardingKey())
+ .setWorkToken(workC.getWorkItem().getWorkToken())
+ .setCacheToken(workC.getWorkItem().getCacheToken())
+ .build())
+ .build();
+
+ Commit commit =
+ Commit.createMultiKey(
+ multiKeyRequest,
+ createComputationState("computationId"),
+ ImmutableList.of(workA, workB, workC));
+
+ workCommitter.start();
+ workCommitter.commit(commit);
+
+ // The entire batch must be aborted immediately without making network calls
+ waitForExpectedSetSize(completeCommits, 3);
+
+ // Verify all three works are aborted individually
+ assertThat(completeCommits)
+ .containsExactly(
+ CompleteCommit.create(
+ "computationId", workA.getShardedKey(), workA.id(), CommitStatus.ABORTED),
+ CompleteCommit.create(
+ "computationId", workB.getShardedKey(), workB.id(), CommitStatus.ABORTED),
+ CompleteCommit.create(
+ "computationId", workC.getShardedKey(), workC.id(), CommitStatus.ABORTED));
+
+ workCommitter.stop();
+ }
+
+ @Test
+ public void testCommit_multiKeyCommitSuccess() {
Review Comment:
nit: put this before the failed test
--
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
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #38768:
URL: https://github.com/apache/beam/pull/38768#discussion_r3634723374
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitter.java:
##########
@@ -147,8 +148,12 @@ private void drainCommitQueue() {
}
private void failCommit(Commit commit) {
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on PR #38768:
URL: https://github.com/apache/beam/pull/38768#issuecomment-4602008017
R: @scwhittle
--
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
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "derrickaw (via GitHub)" <gi...@apache.org>.
derrickaw commented on PR #38768:
URL: https://github.com/apache/beam/pull/38768#issuecomment-4916054087
Hi @scwhittle, can you take a look at your convenience? Thanks
--
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
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #38768:
URL: https://github.com/apache/beam/pull/38768#discussion_r3636726827
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/FakeWindmillServer.java:
##########
@@ -445,6 +460,37 @@ public void flush() {
.orElse(Windmill.CommitStatus.OK));
}
requests.clear();
+
+ for (MultiKeyRequestAndDone elem : multiKeyRequests) {
+ if (dropStreamingCommits) {
+ for (WorkItemCommitRequest workRequest : elem.request.getRequestsList()) {
+ droppedStreamingCommits.put(workRequest.getWorkToken(), elem.onDone);
+ }
+ continue;
+ }
+
+ multiKeyCommitsReceived.add(elem.request);
+ for (WorkItemCommitRequest workRequest : elem.request.getRequestsList()) {
+ commitsReceived.put(workRequest.getWorkToken(), workRequest);
+ }
+
+ // Determine status for the batch.
+ // Default to OK, but if any of the works in the batch has an offered status, use it.
+ Windmill.CommitStatus status = Windmill.CommitStatus.OK;
+ for (WorkItemCommitRequest workRequest : elem.request.getRequestsList()) {
+ Windmill.CommitStatus offeredStatus =
+ streamingCommitsToOffer.remove(
Review Comment:
Good idea, 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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #38768:
URL: https://github.com/apache/beam/pull/38768#discussion_r3634686426
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/Commit.java:
##########
@@ -32,20 +35,43 @@ public abstract class Commit {
public static Commit create(
WorkItemCommitRequest request, ComputationState computationState, Work work) {
Preconditions.checkArgument(request.getSerializedSize() > 0);
- return new AutoValue_Commit(request, computationState, work);
+ return new AutoValue_Commit(
+ Optional.of(request), computationState, Optional.empty(), ImmutableList.of(work));
+ }
+
+ public static Commit createMultiKey(
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest,
+ ComputationState computationState,
+ ImmutableList<Work> workBatch) {
+ Preconditions.checkArgument(!workBatch.isEmpty());
+ return new AutoValue_Commit(
+ Optional.empty(), computationState, Optional.of(multiKeyRequest), workBatch);
}
public final String computationId() {
return computationState().getComputationId();
}
- public abstract WorkItemCommitRequest request();
+ public abstract Optional<WorkItemCommitRequest> singleKeyRequest();
Review Comment:
done.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitterTest.java:
##########
@@ -474,4 +505,201 @@ public void testStop_drainsCommitQueue_concurrentCommit()
waitForExpectedSetSize(completeCommits, sentCommits.intValue());
}
+
+ @Test
+ public void testCommit_multiKeyCommitFailedWork() {
+ Set<CompleteCommit> completeCommits = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ workCommitter = createWorkCommitter(completeCommits::add);
+
+ Work workA = createMockWork(101L);
+ Work workB = createMockWork(102L);
+ Work workC = createMockWork(103L);
+
+ // Mark non-primary key B as failed
+ workB.setFailed();
+
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest =
+ Windmill.MultiKeyWorkItemCommitRequest.newBuilder()
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workA.getWorkItem().getKey())
+ .setShardingKey(workA.getWorkItem().getShardingKey())
+ .setWorkToken(workA.getWorkItem().getWorkToken())
+ .setCacheToken(workA.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workB.getWorkItem().getKey())
+ .setShardingKey(workB.getWorkItem().getShardingKey())
+ .setWorkToken(workB.getWorkItem().getWorkToken())
+ .setCacheToken(workB.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workC.getWorkItem().getKey())
+ .setShardingKey(workC.getWorkItem().getShardingKey())
+ .setWorkToken(workC.getWorkItem().getWorkToken())
+ .setCacheToken(workC.getWorkItem().getCacheToken())
+ .build())
+ .build();
+
+ Commit commit =
+ Commit.createMultiKey(
+ multiKeyRequest,
+ createComputationState("computationId"),
+ ImmutableList.of(workA, workB, workC));
+
+ workCommitter.start();
+ workCommitter.commit(commit);
+
+ // The entire batch must be aborted immediately without making network calls
+ waitForExpectedSetSize(completeCommits, 3);
+
+ // Verify all three works are aborted individually
+ assertThat(completeCommits)
+ .containsExactly(
+ CompleteCommit.create(
+ "computationId", workA.getShardedKey(), workA.id(), CommitStatus.ABORTED),
+ CompleteCommit.create(
+ "computationId", workB.getShardedKey(), workB.id(), CommitStatus.ABORTED),
+ CompleteCommit.create(
+ "computationId", workC.getShardedKey(), workC.id(), CommitStatus.ABORTED));
+
+ workCommitter.stop();
+ }
+
+ @Test
+ public void testCommit_multiKeyCommitSuccess() {
+ Set<CompleteCommit> completeCommits = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ workCommitter = createWorkCommitter(completeCommits::add);
+
+ Work workA = createMockWork(101L);
+ Work workB = createMockWork(102L);
+ Work workC = createMockWork(103L);
+
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest =
+ Windmill.MultiKeyWorkItemCommitRequest.newBuilder()
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workA.getWorkItem().getKey())
+ .setShardingKey(workA.getWorkItem().getShardingKey())
+ .setWorkToken(workA.getWorkItem().getWorkToken())
+ .setCacheToken(workA.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workB.getWorkItem().getKey())
+ .setShardingKey(workB.getWorkItem().getShardingKey())
+ .setWorkToken(workB.getWorkItem().getWorkToken())
+ .setCacheToken(workB.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workC.getWorkItem().getKey())
+ .setShardingKey(workC.getWorkItem().getShardingKey())
+ .setWorkToken(workC.getWorkItem().getWorkToken())
+ .setCacheToken(workC.getWorkItem().getCacheToken())
+ .build())
+ .build();
+
+ Commit commit =
+ Commit.createMultiKey(
+ multiKeyRequest,
+ createComputationState("computationId"),
+ ImmutableList.of(workA, workB, workC));
+
+ workCommitter.start();
+ workCommitter.commit(commit);
+
+ // Wait for the server to receive and process the commits
+ fakeWindmillServer.waitForAndGetCommits(3);
+ waitForExpectedSetSize(completeCommits, 3);
+
+ // Verify that FakeWindmillServer received all 3 work requests in multiKeyCommitsReceived
+ List<Windmill.MultiKeyWorkItemCommitRequest> multiKeyCommits =
+ fakeWindmillServer.getMultiKeyCommitsReceived();
+ assertThat(multiKeyCommits).hasSize(1);
+ assertThat(multiKeyCommits.get(0)).isEqualTo(multiKeyRequest);
+
+ // Verify all three works are completed successfully
+ assertThat(completeCommits)
+ .containsExactly(
+ CompleteCommit.create(
+ "computationId", workA.getShardedKey(), workA.id(), CommitStatus.OK),
+ CompleteCommit.create(
+ "computationId", workB.getShardedKey(), workB.id(), CommitStatus.OK),
+ CompleteCommit.create(
+ "computationId", workC.getShardedKey(), workC.id(), CommitStatus.OK));
+
+ workCommitter.stop();
+ }
+
+ @Test
+ public void testCommit_multiKeyCommitStatusNotOK() {
+ Set<CompleteCommit> completeCommits = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ workCommitter = createWorkCommitter(completeCommits::add);
+
+ Work workA = createMockWork(101L);
+ Work workB = createMockWork(102L);
+ Work workC = createMockWork(103L);
+
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest =
+ Windmill.MultiKeyWorkItemCommitRequest.newBuilder()
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workA.getWorkItem().getKey())
+ .setShardingKey(workA.getWorkItem().getShardingKey())
+ .setWorkToken(workA.getWorkItem().getWorkToken())
+ .setCacheToken(workA.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workB.getWorkItem().getKey())
+ .setShardingKey(workB.getWorkItem().getShardingKey())
+ .setWorkToken(workB.getWorkItem().getWorkToken())
+ .setCacheToken(workB.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workC.getWorkItem().getKey())
+ .setShardingKey(workC.getWorkItem().getShardingKey())
+ .setWorkToken(workC.getWorkItem().getWorkToken())
+ .setCacheToken(workC.getWorkItem().getCacheToken())
+ .build())
+ .build();
+
+ Commit commit =
+ Commit.createMultiKey(
+ multiKeyRequest,
+ createComputationState("computationId"),
+ ImmutableList.of(workA, workB, workC));
+
+ // Offer NOT_FOUND status for one of the works.
+ fakeWindmillServer.whenCommitWorkStreamCalled().put(workB.id(), CommitStatus.NOT_FOUND);
+
+ workCommitter.start();
+ workCommitter.commit(commit);
+
+ // Wait for the server to receive and process the commits
+ fakeWindmillServer.waitForAndGetCommits(3);
+ waitForExpectedSetSize(completeCommits, 3);
+
+ // Verify that FakeWindmillServer received the multi-key commit
+ List<Windmill.MultiKeyWorkItemCommitRequest> multiKeyCommits =
+ fakeWindmillServer.getMultiKeyCommitsReceived();
+ assertThat(multiKeyCommits).hasSize(1);
+ assertThat(multiKeyCommits.get(0)).isEqualTo(multiKeyRequest);
+
+ // Verify all three works in the multi-key commit are completed with NOT_FOUND status
Review Comment:
Added a check for `currentActiveCommitBytes() == 0` that checks commits are not pending in commiter anymore.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitterTest.java:
##########
@@ -186,10 +186,14 @@ public void testCommit_sendsCommitsToStreamingEngine() {
waitForExpectedSetSize(completeCommits, 5);
for (Commit commit : commits) {
- WorkItemCommitRequest request = committed.get(commit.work().getWorkItem().getWorkToken());
+ WorkItemCommitRequest request =
+ committed.get(commit.workBatch().get(0).getWorkItem().getWorkToken());
Review Comment:
done.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingApplianceWorkCommitterTest.java:
##########
@@ -129,9 +129,9 @@ public void testCommit() {
for (Commit commit : commits) {
Windmill.WorkItemCommitRequest request =
- committed.get(commit.work().getWorkItem().getWorkToken());
+ committed.get(commit.workBatch().get(0).getWorkItem().getWorkToken());
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #38768:
URL: https://github.com/apache/beam/pull/38768#discussion_r3636562704
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitterTest.java:
##########
@@ -224,14 +228,24 @@ public void testCommit_handlesFailedCommits() {
waitForExpectedSetSize(completeCommits, 10);
for (Commit commit : commits) {
- if (commit.work().isFailed()) {
+ if (commit.isFailed()) {
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #38768:
URL: https://github.com/apache/beam/pull/38768#issuecomment-4602009297
## Summary of Changes
Hello, I'm Gemini Code Assist[^1]! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request introduces support for MultiKey commits in the Dataflow streaming engine. By defining a new MultiKeyWorkItemCommitRequest and updating the commit orchestration logic, the system can now batch multiple work items into a single commit request, significantly improving efficiency and throughput for streaming pipelines.
### Highlights
* **Proto Definitions**: Added MultiKeyWorkItemCommitRequest and Uint128Proto to windmill.proto to support batching multiple work items.
* **Commit Model Updates**: Updated the Commit class to handle multi-key requests and support batching multiple work items.
* **Committer Logic**: Refactored StreamingEngineWorkCommitter to process multi-key commits and manage work batches effectively.
* **gRPC Streaming**: Updated GrpcCommitWorkStream to support COMMIT_TYPE_MULTI_KEY and ensure efficient streaming of batched requests.
* **Testing**: Added comprehensive tests in FakeWindmillServer and updated existing unit tests to verify multi-key commit functionality.
<details>
<summary><b>New Features</b></summary>
🧠You can now enable **Memory** (public preview) to help **Gemini Code Assist** learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. **Click [here](https://codeassist.google/code-review/login) to enable Memory in your admin console.**
</details>
<details>
<summary><b>Using Gemini Code Assist</b></summary>
<br>
The full guide for Gemini Code Assist can be found on our [documentation page](https://developers.google.com/gemini-code-assist/docs/review-github-code), here are some quick tips.
<b>Invoking Gemini</b>
You can request assistance from Gemini at any point by creating a comment using either `/gemini <command>` or `@gemini-code-assist <command>`. Below is a summary of the supported commands on the current page.
Feature | Command | Description
--- | --- | ---
Code Review | `/gemini review` | Performs a code review for the current pull request in its current state.
Pull Request Summary | `/gemini summary` | Provides a summary of the current pull request in its current state.
Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help | `/gemini help` | Displays a list of available commands.
<b>Customization</b>
To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a `.gemini/` folder in the base of the repository. Detailed instructions can be found [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github).
<b>Limitations & Feedback</b>
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with :thumbsup: and :thumbsdown: on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up [here](https://google.qualtrics.com/jfe/form/SV_2cyuGuTWsEw84yG).
</details>
[^1]: Review the [Privacy Notices](https://policies.google.com/privacy), [Generative AI Prohibited Use Policy](https://policies.google.com/terms/generative-ai/use-policy), [Terms of Service](https://policies.google.com/terms), and learn how to configure Gemini Code Assist in GitHub [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). Gemini can make mistakes, so double check it and [use code with caution](https://support.google.com/legal/answer/13505487).
--
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
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #38768:
URL: https://github.com/apache/beam/pull/38768#discussion_r3636569497
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStreamTest.java:
##########
@@ -1134,6 +1134,76 @@ public void testCommitWorkItem_multiplePhysicalStreams_multipleHandovers_halfClo
assertTrue(commitWorkStream.awaitTermination(10, TimeUnit.SECONDS));
}
+ @Test
+ public void testCommit_multiKeyCommit() throws Exception {
+ GrpcCommitWorkStream commitWorkStream = createCommitWorkStream();
+ FakeWindmillGrpcService.CommitStreamInfo streamInfo = waitForConnectionAndConsumeHeader();
+
+ CompletableFuture<Windmill.CommitStatus> commitStatusFuture = new CompletableFuture<>();
+
+ // 1. Construct two individual WorkItemCommitRequests
+ long shardingKey1 = 101L;
+ long workToken1 = 201L;
+ long cacheToken1 = 301L;
+ long shardingKey2 = 102L;
+ long workToken2 = 202L;
+ long cacheToken2 = 302L;
+ Windmill.WorkItemCommitRequest request1 =
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(ByteString.copyFromUtf8("key1"))
+ .setShardingKey(shardingKey1)
+ .setWorkToken(workToken1)
+ .setCacheToken(cacheToken1)
+ .build();
+ Windmill.WorkItemCommitRequest request2 =
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(ByteString.copyFromUtf8("key2"))
+ .setShardingKey(shardingKey2)
+ .setWorkToken(workToken2)
+ .setCacheToken(cacheToken2)
+ .build();
+
+ // 2. Wrap them into a MultiKeyWorkItemCommitRequest
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest =
+ Windmill.MultiKeyWorkItemCommitRequest.newBuilder()
+ .addRequests(request1)
+ .addRequests(request2)
+ .build();
+
+ // 3. Commit the multi-key work item using the request batcher
+ try (WindmillStream.CommitWorkStream.RequestBatcher batcher = commitWorkStream.batcher()) {
+ assertTrue(
+ batcher.commitMultiKeyWorkItem(
+ COMPUTATION_ID, multiKeyRequest, commitStatusFuture::complete));
+ }
+
+ // 4. Receive and assert request properties on FakeWindmillGrpcService
+ Windmill.StreamingCommitWorkRequest request = streamInfo.requests.take();
+ assertThat(request.getCommitChunkCount()).isEqualTo(1);
+
+ Windmill.StreamingCommitRequestChunk chunk = request.getCommitChunk(0);
+
+ // Assert that the commit type is correctly identified as COMMIT_TYPE_MULTI_KEY
+ assertThat(chunk.getCommitType())
+ .isEqualTo(Windmill.StreamingCommitRequestChunk.CommitType.COMMIT_TYPE_MULTI_KEY);
+
+ // Assert that the routing sharding key is mapped to the first request's sharding key
+ assertThat(chunk.getShardingKey()).isEqualTo(request1.getShardingKey());
+
+ // Assert that the serialized payload matches the input multiKeyRequest
+ Windmill.MultiKeyWorkItemCommitRequest parsedRequest =
+ Windmill.MultiKeyWorkItemCommitRequest.parseFrom(chunk.getSerializedWorkItemCommit());
+ assertThat(parsedRequest).isEqualTo(multiKeyRequest);
+
+ // 5. Respond with the generated requestId to complete the commit
+ long requestId = chunk.getRequestId();
+ streamInfo.responseObserver.onNext(
+ Windmill.StreamingCommitResponse.newBuilder().addRequestId(requestId).build());
+
+ // 6. Verify callback completed successfully with CommitStatus.OK
+ assertThat(commitStatusFuture.get()).isEqualTo(Windmill.CommitStatus.OK);
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #38768:
URL: https://github.com/apache/beam/pull/38768#discussion_r3634723125
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitter.java:
##########
@@ -113,8 +114,8 @@ public void commit(Commit commit) {
"Trying to queue commit on shutdown, failing commit=[computationId={}, shardingKey={},"
+ " workId={} ].",
commit.computationId(),
- commit.work().getShardedKey(),
- commit.work().id());
+ commit.workBatch().get(0).getShardedKey(),
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #38768:
URL: https://github.com/apache/beam/pull/38768
--
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
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #38768:
URL: https://github.com/apache/beam/pull/38768#discussion_r3340876671
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -378,34 +382,34 @@ private void issueMultiChunkRequest(long id, PendingRequest pendingRequest)
@AutoValue
abstract static class PendingRequest {
-
- private static PendingRequest create(
- String computationId, WorkItemCommitRequest request, Consumer<CommitStatus> onDone) {
- return new AutoValue_GrpcCommitWorkStream_PendingRequest(computationId, request, onDone);
+ static PendingRequest create(
+ String computationId,
+ long shardingKey,
+ ByteString serializedCommit,
+ StreamingCommitRequestChunk.CommitType commitType,
+ Consumer<CommitStatus> onDone) {
+ return new AutoValue_GrpcCommitWorkStream_PendingRequest(
+ computationId, shardingKey, serializedCommit, commitType, onDone);
}
abstract String computationId();
- abstract WorkItemCommitRequest request();
+ abstract long shardingKey();
+
+ abstract ByteString serializedCommit();
+
+ abstract StreamingCommitRequestChunk.CommitType commitType();
abstract Consumer<CommitStatus> onDone();
private long getBytes() {
- return (long) request().getSerializedSize() + computationId().length();
- }
-
- private ByteString serializedCommit() {
- return request().toByteString();
+ return (long) serializedCommit().size() + computationId().length();
}
Review Comment:
### Performance Optimization: Defer Serialization to the Stream Writer Thread
Currently, `PendingRequest` eagerly serializes the commit request to a `ByteString` on the committer thread (inside `commitWorkItem` and `commitMultiKeyWorkItem`). Eager serialization on the committer thread can become a performance bottleneck under high throughput.
By holding the `MessageLite` (or the vendored equivalent `org.apache.beam.vendor.grpc.v1p69p0.com.google.protobuf.MessageLite`) directly in `PendingRequest` and using `@AutoValue.Memoized` to lazily serialize it, we can:
1. Defer the serialization overhead to the stream writer thread (when `serializedCommit()` is actually called).
2. Ensure that serialization is only performed once and cached, even if `serializedCommit()` is accessed multiple times.
```java
static PendingRequest create(
String computationId,
long shardingKey,
com.google.protobuf.MessageLite request,
StreamingCommitRequestChunk.CommitType commitType,
Consumer<CommitStatus> onDone) {
return new AutoValue_GrpcCommitWorkStream_PendingRequest(
computationId, shardingKey, request, commitType, onDone);
}
abstract String computationId();
abstract long shardingKey();
abstract com.google.protobuf.MessageLite request();
abstract StreamingCommitRequestChunk.CommitType commitType();
abstract Consumer<CommitStatus> onDone();
private long getBytes() {
return (long) serializedCommit().size() + computationId().length();
}
@AutoValue.Memoized
ByteString serializedCommit() {
return request().toByteString();
}
```
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -462,7 +466,34 @@ public boolean commitWorkItem(
return false;
}
- PendingRequest request = PendingRequest.create(computation, commitRequest, onDone);
+ PendingRequest request =
+ PendingRequest.create(
+ computation,
+ commitRequest.getShardingKey(),
+ commitRequest.toByteString(),
+ StreamingCommitRequestChunk.CommitType.COMMIT_TYPE_SINGLE_KEY,
+ onDone);
+ add(idGenerator.incrementAndGet(), request);
+ return true;
+ }
+
+ @Override
+ public boolean commitMultiKeyWorkItem(
+ String computation,
+ Windmill.MultiKeyWorkItemCommitRequest commitRequest,
+ Consumer<CommitStatus> onDone) {
+ if (!canAccept(commitRequest.getSerializedSize() + computation.length())) {
+ return false;
+ }
+ Preconditions.checkArgument(commitRequest.getRequestsCount() > 0);
+ PendingRequest request =
+ PendingRequest.create(
+ computation,
+ // Any key in the batch for routing
+ commitRequest.getRequests(0).getShardingKey(),
+ commitRequest.toByteString(),
+ StreamingCommitRequestChunk.CommitType.COMMIT_TYPE_MULTI_KEY,
+ onDone);
Review Comment:
Pass the `commitRequest` directly to `PendingRequest.create` to support lazy serialization.
```suggestion
PendingRequest request =
PendingRequest.create(
computation,
// Any key in the batch for routing
commitRequest.getRequests(0).getShardingKey(),
commitRequest,
StreamingCommitRequestChunk.CommitType.COMMIT_TYPE_MULTI_KEY,
onDone);
```
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -462,7 +466,34 @@ public boolean commitWorkItem(
return false;
}
- PendingRequest request = PendingRequest.create(computation, commitRequest, onDone);
+ PendingRequest request =
+ PendingRequest.create(
+ computation,
+ commitRequest.getShardingKey(),
+ commitRequest.toByteString(),
+ StreamingCommitRequestChunk.CommitType.COMMIT_TYPE_SINGLE_KEY,
+ onDone);
Review Comment:
Pass the `commitRequest` directly to `PendingRequest.create` to support lazy serialization.
```suggestion
PendingRequest request =
PendingRequest.create(
computation,
commitRequest.getShardingKey(),
commitRequest,
StreamingCommitRequestChunk.CommitType.COMMIT_TYPE_SINGLE_KEY,
onDone);
```
--
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
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38768:
URL: https://github.com/apache/beam/pull/38768#issuecomment-4602016705
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers`
--
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
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #38768:
URL: https://github.com/apache/beam/pull/38768#discussion_r3636728850
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/FakeWindmillServer.java:
##########
@@ -423,6 +426,18 @@ public boolean commitWorkItem(
return true;
}
+ @Override
+ public boolean commitMultiKeyWorkItem(
+ String computation,
+ Windmill.MultiKeyWorkItemCommitRequest request,
+ Consumer<Windmill.CommitStatus> onDone) {
+ LOG.debug("commitWorkStream::commitMultiKeyWorkItem: {}", request);
+ if (multiKeyRequests.size() > 5) return false;
Review Comment:
thanks, removed.
--
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
Re: [PR] [Dataflow Streaming][Multikey] Support MultiKey commits in windmill clients [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #38768:
URL: https://github.com/apache/beam/pull/38768#discussion_r3634725420
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -512,7 +528,34 @@ public boolean commitWorkItem(
return false;
}
- PendingRequest request = new PendingRequest(computation, commitRequest, onDone);
+ PendingRequest request =
+ new PendingRequest(
+ computation,
+ commitRequest.getShardingKey(),
+ commitRequest.toByteString(),
+ StreamingCommitRequestChunk.CommitType.COMMIT_TYPE_SINGLE_KEY,
+ onDone);
+ add(idGenerator.incrementAndGet(), request);
+ return true;
+ }
+
+ @Override
+ public boolean commitMultiKeyWorkItem(
+ String computation,
+ Windmill.MultiKeyWorkItemCommitRequest commitRequest,
+ Consumer<CommitStatus> onDone) {
+ if (!canAccept(commitRequest.getSerializedSize() + computation.length())) {
+ return false;
+ }
+ Preconditions.checkArgument(commitRequest.getRequestsCount() > 0);
Review Comment:
done.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/Commit.java:
##########
@@ -32,20 +35,43 @@ public abstract class Commit {
public static Commit create(
WorkItemCommitRequest request, ComputationState computationState, Work work) {
Preconditions.checkArgument(request.getSerializedSize() > 0);
- return new AutoValue_Commit(request, computationState, work);
+ return new AutoValue_Commit(
+ Optional.of(request), computationState, Optional.empty(), ImmutableList.of(work));
+ }
+
+ public static Commit createMultiKey(
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest,
+ ComputationState computationState,
+ ImmutableList<Work> workBatch) {
+ Preconditions.checkArgument(!workBatch.isEmpty());
+ return new AutoValue_Commit(
+ Optional.empty(), computationState, Optional.of(multiKeyRequest), workBatch);
}
public final String computationId() {
return computationState().getComputationId();
}
- public abstract WorkItemCommitRequest request();
+ public abstract Optional<WorkItemCommitRequest> singleKeyRequest();
public abstract ComputationState computationState();
- public abstract Work work();
+ public abstract Optional<Windmill.MultiKeyWorkItemCommitRequest> multiKeyRequest();
+
+ public abstract ImmutableList<Work> workBatch();
+
+ public final boolean isFailed() {
+ for (Work w : workBatch()) {
+ if (w.isFailed()) {
+ return true;
+ }
+ }
+ return false;
+ }
public final int getSize() {
Review Comment:
done.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitterTest.java:
##########
@@ -282,11 +296,16 @@ public void testCommit_handlesCompleteCommits_commitStatusNotOK() {
waitForExpectedSetSize(completeCommits, commits.size());
for (Commit commit : commits) {
- WorkItemCommitRequest request = committed.get(commit.work().getWorkItem().getWorkToken());
+ WorkItemCommitRequest request =
Review Comment:
done
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingApplianceWorkCommitterTest.java:
##########
@@ -141,12 +141,13 @@ public void testCommit() {
(CompleteCommit completeCommit, Commit commit) ->
completeCommit.computationId().equals(commit.computationId())
&& completeCommit.status() == Windmill.CommitStatus.OK
- && completeCommit.workId().equals(commit.work().id())
+ && completeCommit.workId().equals(commit.workBatch().get(0).id())
Review Comment:
done.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitterTest.java:
##########
@@ -409,10 +436,14 @@ public void testMultipleCommitSendersSingleStream() {
waitForExpectedSetSize(completeCommits, commits.size());
for (Commit commit : commits) {
- WorkItemCommitRequest request = committed.get(commit.work().getWorkItem().getWorkToken());
+ WorkItemCommitRequest request =
Review Comment:
done.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitterTest.java:
##########
@@ -474,4 +505,201 @@ public void testStop_drainsCommitQueue_concurrentCommit()
waitForExpectedSetSize(completeCommits, sentCommits.intValue());
}
+
+ @Test
+ public void testCommit_multiKeyCommitFailedWork() {
+ Set<CompleteCommit> completeCommits = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ workCommitter = createWorkCommitter(completeCommits::add);
+
+ Work workA = createMockWork(101L);
+ Work workB = createMockWork(102L);
+ Work workC = createMockWork(103L);
+
+ // Mark non-primary key B as failed
+ workB.setFailed();
+
+ Windmill.MultiKeyWorkItemCommitRequest multiKeyRequest =
+ Windmill.MultiKeyWorkItemCommitRequest.newBuilder()
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workA.getWorkItem().getKey())
+ .setShardingKey(workA.getWorkItem().getShardingKey())
+ .setWorkToken(workA.getWorkItem().getWorkToken())
+ .setCacheToken(workA.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workB.getWorkItem().getKey())
+ .setShardingKey(workB.getWorkItem().getShardingKey())
+ .setWorkToken(workB.getWorkItem().getWorkToken())
+ .setCacheToken(workB.getWorkItem().getCacheToken())
+ .build())
+ .addRequests(
+ Windmill.WorkItemCommitRequest.newBuilder()
+ .setKey(workC.getWorkItem().getKey())
+ .setShardingKey(workC.getWorkItem().getShardingKey())
+ .setWorkToken(workC.getWorkItem().getWorkToken())
+ .setCacheToken(workC.getWorkItem().getCacheToken())
+ .build())
+ .build();
+
+ Commit commit =
+ Commit.createMultiKey(
+ multiKeyRequest,
+ createComputationState("computationId"),
+ ImmutableList.of(workA, workB, workC));
+
+ workCommitter.start();
+ workCommitter.commit(commit);
+
+ // The entire batch must be aborted immediately without making network calls
+ waitForExpectedSetSize(completeCommits, 3);
+
+ // Verify all three works are aborted individually
+ assertThat(completeCommits)
+ .containsExactly(
+ CompleteCommit.create(
+ "computationId", workA.getShardedKey(), workA.id(), CommitStatus.ABORTED),
+ CompleteCommit.create(
+ "computationId", workB.getShardedKey(), workB.id(), CommitStatus.ABORTED),
+ CompleteCommit.create(
+ "computationId", workC.getShardedKey(), workC.id(), CommitStatus.ABORTED));
+
+ workCommitter.stop();
+ }
+
+ @Test
+ public void testCommit_multiKeyCommitSuccess() {
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org