fapaul commented on code in PR #25353:
URL: https://github.com/apache/flink/pull/25353#discussion_r1775075667
##########
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 you please still fix it?
##########
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:
I am not sure about the location you added the new assertions. You have
added them `SinkV1TransformationTranslatorITCase` while I would have expected
the assertions in `SinkTransformationTranslatorITCaseBase`
--
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]
