fapaul commented on code in PR #25353:
URL: https://github.com/apache/flink/pull/25353#discussion_r1771171220
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollector.java:
##########
@@ -254,4 +249,9 @@ private CheckpointCommittableManagerImpl<CommT>
getCheckpointCommittables(
this.checkpointCommittables.get(committable.getCheckpointIdOrEOI());
return checkNotNull(committables, "Unknown checkpoint for %s",
committable);
}
+
+ /** Removes all metadata about checkpoints of which all committables are
fully committed. */
+ public void cleanFinished() {
Review Comment:
Why do we need this method? I liked the previous approach, where the
collector cleaned up after retrieval. With this change, the responsibility has
moved to the caller, leaving the risk of not cleaning up at all in case
`cleanFinished` is never called.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/partitioner/StreamPartitioner.java:
##########
@@ -34,6 +34,13 @@ public abstract class StreamPartitioner<T>
protected int numberOfChannels;
+ /**
+ * By default, all partitioner except {@link #isBroadcast()} or {@link
#isPointwise()} support
+ * unaligned checkpoints. However, transformations may disable unaligned
checkpoints for
+ * specific edges.
+ */
+ protected boolean supportsUnalignedCheckpoint = true;
Review Comment:
Can this be `private`?
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java:
##########
@@ -185,6 +188,58 @@ private void expand() {
}
}
+ private List<Transformation<?>> getSinkTransformations(int sizeBefore)
{
+ return executionEnvironment
+ .getTransformations()
+ .subList(sizeBefore,
executionEnvironment.getTransformations().size());
+ }
+
+ /**
+ * Disables UC for all connections of operators within the sink
expansion. This is necessary
+ * because committables need to be at the respective operators on
notifyCheckpointComplete
+ * or else we can't commit all side-effects, which violates the
contract of
+ * notifyCheckpointComplete.
+ */
+ private void disallowUnalignedCheckpoint(List<Transformation<?>>
sinkTransformations) {
Review Comment:
Can you add a test in `SinkV2TransformationTranslatorITCase` for this?
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java:
##########
@@ -185,6 +188,58 @@ private void expand() {
}
}
+ private List<Transformation<?>> getSinkTransformations(int sizeBefore)
{
+ return executionEnvironment
+ .getTransformations()
+ .subList(sizeBefore,
executionEnvironment.getTransformations().size());
+ }
+
+ /**
+ * Disables UC for all connections of operators within the sink
expansion. This is necessary
+ * because committables need to be at the respective operators on
notifyCheckpointComplete
+ * or else we can't commit all side-effects, which violates the
contract of
+ * notifyCheckpointComplete.
+ */
+ private void disallowUnalignedCheckpoint(List<Transformation<?>>
sinkTransformations) {
+ Optional<Transformation<?>> writerOpt =
+
sinkTransformations.stream().filter(SinkExpander::isWriter).findFirst();
+ Preconditions.checkState(writerOpt.isPresent(), "Writer
transformation not found.");
+ Transformation<?> writer = writerOpt.get();
+ int indexOfWriter = sinkTransformations.indexOf(writer);
+
+ // check all transformation after the writer and recursively
disable UC for all inputs
Review Comment:
Nit: `recursively` -> `iteratively`
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorTestBase.java:
##########
@@ -126,42 +127,6 @@ void
testWaitForCommittablesOfLatestCheckpointBeforeCommitting() throws Exceptio
testHarness.close();
}
- @Test
- void testImmediatelyCommitLateCommittables() throws Exception {
Review Comment:
Why was this test 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
