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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
