fapaul commented on code in PR #20979:
URL: https://github.com/apache/flink/pull/20979#discussion_r990999206
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java:
##########
@@ -53,10 +54,12 @@
CheckpointCommittableManagerImpl(
Map<Integer, SubtaskCommittableManager<CommT>>
subtasksCommittableManagers,
+ int subtaskId,
+ int numberOfSubtasks,
@Nullable Long checkpointId) {
this.subtasksCommittableManagers =
checkNotNull(subtasksCommittableManagers);
- this.subtaskId = 0;
- this.numberOfSubtasks = 1;
+ this.subtaskId = subtaskId;
Review Comment:
Nit: can you call the other ctor here to keep it consistent?
```java
this(subtaskId, numberOfSubtaks, checkpointId)
```
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializerTest.java:
##########
@@ -72,38 +73,86 @@ void testCommittableCollectorV1SerDe() throws IOException {
@Test
void testCommittableCollectorV2SerDe() throws IOException {
- final CommittableCollector<Integer> committableCollector = new
CommittableCollector<>(2, 3);
- committableCollector.addMessage(new CommittableSummary<>(2, 3, 1L, 1,
1, 0));
- committableCollector.addMessage(new CommittableSummary<>(2, 3, 2L, 1,
1, 0));
- committableCollector.addMessage(new CommittableWithLineage<>(1, 1L,
2));
- committableCollector.addMessage(new CommittableWithLineage<>(2, 2L,
2));
+ int subtaskId = 2;
Review Comment:
I think it would be better to add/adapt a test in the
`CommitterOperatorTest`. You already have to expose more internals i.e.
`@VisibleForTesting` in the CheckpointCommittableManagerImpl` which is a
somewhat bad sign.
In the end, we want to have a test where, a `CommitterOperater` with a
subtaskId != 0 recovers the committables and sends them forward and we want to
assert that the subtaskId is the one from the new `CommitterOperator`
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java:
##########
@@ -158,6 +161,18 @@ CheckpointCommittableManagerImpl<CommT> copy() {
return new CheckpointCommittableManagerImpl<>(
subtasksCommittableManagers.entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey, (e) ->
e.getValue().copy())),
+ subtaskId,
Review Comment:
Can you add a simple test that during copy, the subtaskId and
numberOfSubtasks are preserved?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java:
##########
@@ -158,6 +161,18 @@ CheckpointCommittableManagerImpl<CommT> copy() {
return new CheckpointCommittableManagerImpl<>(
subtasksCommittableManagers.entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey, (e) ->
e.getValue().copy())),
+ subtaskId,
+ numberOfSubtasks,
checkpointId);
}
+
+ @VisibleForTesting
Review Comment:
You can use `getSummary` to receive this information.
--
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]
