[FLINK-36788] Fix and cover global committer #25685
Conversation
Small refactor for next commit
StandardSinkTopologies didn't expose yet the newly added ctx, such that the global committer couldn't access the metric groups
...ime/src/main/java/org/apache/flink/streaming/api/connector/sink2/StandardSinkTopologies.java
Show resolved
Hide resolved
| @@ -90,7 +92,18 @@ private Collection<Integer> translateInternal( | |||
| transformation.setName(GLOBAL_COMMITTER_TRANSFORMATION_NAME); | |||
| transformation.setParallelism(1); | |||
| transformation.setMaxParallelism(1); | |||
| return Collections.emptyList(); | |||
| copySafely(transformation::setName, globalCommitterTransform::getName); | |||
There was a problem hiding this comment.
In which scenarios does the globalCommitterTransform have it's own attributes?
There was a problem hiding this comment.
We set the attributes while expanding the post commit topology. We expand this transform almost at the end of the sink expansion. So it's fair to say that we have it in all cases where we set the properties during post commit expansion. It's not set if no uid or customUidHashes are user-supplied. I think name should always be set but I still use the copySafely method because you never know how things will evolve.
...c/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java
Show resolved
Hide resolved
...src/test/java/org/apache/flink/streaming/api/graph/SinkV2TransformationTranslatorITCase.java
Show resolved
Hide resolved
...reaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/TestSinkV2.java
Show resolved
Hide resolved
They are only semi-related. Basically while adding the global committer to TestSinkV2, I noticed that we lack the ctx to do so. I could pull them out into a separate PR but I also feel like these changes are minor enough to piggy-back (it's less than 50 LOC on the first two commits). |
So far, the global committer didn't properly get all properties set during expansion. In some cases, it didn't get expanded at all. This commit also inline SinkTransformationTranslatorITCaseBase.java after the respective V1 implementation was removed. V2 now has the same coverage as V1 used to have.
AHeise
force-pushed
the
FLINK-36788-cover-global-committer
branch
from
bbf92af
to
960cda3
Compare
|
@flinkbot run azure |
What is the purpose of the change
With the removal of Sink V1, we missed crucial coverage of the global committer coming from the StandardSinkTopologies. That caused a previous PR #25456 to contain undetected bugs.
This PR readds the coverage and fixes the uncovered bugs.
Brief change log
Verifying this change
This change added tests and can be verified as follows:
SinkV2TransformationTranslatorITCase
Does this pull request potentially affect one of the following parts:
@Public(Evolving): (yes / no)Documentation