kristoffSC commented on code in PR #20979:
URL: https://github.com/apache/flink/pull/20979#discussion_r992352496
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializerTest.java:
##########
@@ -72,38 +73,88 @@ 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;
+ int numberOfSubtasks = 3;
+
+ final CommittableCollectorSerializer<Integer> ccSerializer =
+ new CommittableCollectorSerializer<>(
+ COMMITTABLE_SERIALIZER, subtaskId, numberOfSubtasks);
+
+ final CommittableCollector<Integer> committableCollector =
+ new CommittableCollector<>(subtaskId, numberOfSubtasks);
+ committableCollector.addMessage(
+ new CommittableSummary<>(subtaskId, numberOfSubtasks, 1L, 1,
1, 0));
+ committableCollector.addMessage(
+ new CommittableSummary<>(subtaskId, numberOfSubtasks, 2L, 1,
1, 0));
+ committableCollector.addMessage(new CommittableWithLineage<>(1, 1L,
subtaskId));
+ committableCollector.addMessage(new CommittableWithLineage<>(2, 2L,
subtaskId));
+
final CommittableCollector<Integer> copy =
- SERIALIZER.deserialize(2,
SERIALIZER.serialize(committableCollector));
+ ccSerializer.deserialize(2,
SERIALIZER.serialize(committableCollector));
// Expect the subtask Id equal to the origin of the collector
- assertThat(copy.getSubtaskId()).isEqualTo(1);
+ assertThat(copy.getSubtaskId()).isEqualTo(subtaskId);
assertThat(copy.isFinished()).isFalse();
- assertThat(copy.getNumberOfSubtasks()).isEqualTo(1);
- final Collection<CheckpointCommittableManagerImpl<Integer>>
checkpointCommittables =
- committableCollector.getCheckpointCommittables();
- assertThat(checkpointCommittables).hasSize(2);
- final Iterator<CheckpointCommittableManagerImpl<Integer>>
committablesIterator =
- checkpointCommittables.iterator();
- final SubtaskCommittableManager<Integer>
subtaskCommittableManagerCheckpoint1 =
- committablesIterator.next().getSubtaskCommittableManager(2);
- assertThat(
- subtaskCommittableManagerCheckpoint1
- .getPendingRequests()
- .map(CommitRequestImpl::getCommittable)
- .collect(Collectors.toList()))
- .containsExactly(1);
- final SubtaskCommittableManager<Integer>
subtaskCommittableManagerCheckpoint2 =
- committablesIterator.next().getSubtaskCommittableManager(2);
- assertThat(
- subtaskCommittableManagerCheckpoint2
- .getPendingRequests()
- .map(CommitRequestImpl::getCommittable)
- .collect(Collectors.toList()))
- .containsExactly(2);
+ assertThat(copy.getNumberOfSubtasks()).isEqualTo(numberOfSubtasks);
+
+ // assert original CommittableCollector
+ assertCommittableCollector(
+ "Original CommittableCollector", subtaskId, numberOfSubtasks,
committableCollector);
+
+ // assert deserialized CommittableCollector
+ assertCommittableCollector(
+ "Deserialized CommittableCollector", subtaskId,
numberOfSubtasks, copy);
+ }
+
+ private void assertCommittableCollector(
+ String assertMessageHeading,
+ int subtaskId,
+ int numberOfSubtasks,
+ CommittableCollector<Integer> committableCollector) {
+
+ assertAll(
+ assertMessageHeading,
+ () -> {
+ final Collection<CheckpointCommittableManagerImpl<Integer>>
+ checkpointCommittables =
+
committableCollector.getCheckpointCommittables();
+ assertThat(checkpointCommittables).hasSize(2);
+
+ final Iterator<CheckpointCommittableManagerImpl<Integer>>
committablesIterator =
+ checkpointCommittables.iterator();
+ final CheckpointCommittableManagerImpl<Integer>
checkpointCommittableManager1 =
+ committablesIterator.next();
+ final SubtaskCommittableManager<Integer>
subtaskCommittableManagerCheckpoint1 =
+
checkpointCommittableManager1.getSubtaskCommittableManager(subtaskId);
+
assertThat(checkpointCommittableManager1.getSummary().getSubtaskId())
+ .isEqualTo(subtaskId);
+
assertThat(checkpointCommittableManager1.getSummary().getNumberOfSubtasks())
+ .isEqualTo(numberOfSubtasks);
+ assertThat(
+ subtaskCommittableManagerCheckpoint1
+ .getPendingRequests()
+
.map(CommitRequestImpl::getCommittable)
+ .collect(Collectors.toList()))
+ .containsExactly(1);
+
assertThat(subtaskCommittableManagerCheckpoint1.getSubtaskId())
+ .isEqualTo(subtaskId);
+
+ final CheckpointCommittableManagerImpl<Integer>
checkpointCommittableManager2 =
+ committablesIterator.next();
+ final SubtaskCommittableManager<Integer>
subtaskCommittableManagerCheckpoint2 =
+
checkpointCommittableManager2.getSubtaskCommittableManager(subtaskId);
+
assertThat(checkpointCommittableManager2.getSummary().getSubtaskId())
Review Comment:
Fixed, plz take a look.
--
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]
