[
https://issues.apache.org/jira/browse/FLINK-39192?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18137169#comment-18137169
]
Purushottam Sinha commented on FLINK-39192:
-------------------------------------------
Probable similar cause of failure [~martijnvisser]
> SinkWriterOperator in post-commit topology throws IllegalStateException with
> bounded source in streaming mode
> -------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-39192
> URL: https://issues.apache.org/jira/browse/FLINK-39192
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Task
> Affects Versions: 2.1.0, 2.2.0
> Reporter: Calico Shop
> Priority: Major
>
> When a *Sink* implements *SupportsPostCommitTopology* and the job uses a
> *bounded* source in streaming mode ({*}RuntimeExecutionMode.STREAMING{*}),
> the *SinkWriterOperator* wrapping the post-commit sink throws:
> {code:java}
> java.lang.IllegalStateException: Received element after endOfInput{code}
> h3. Root cause
> With a bounded source in streaming mode, the following sequence occurs:
> 1. Source emits all records and finishes
> 2. *endOfInput* propagates through the operator chain; the post-commit
> {*}SinkWriterOperator{*}'s *endInput()* sets *endOfInput = true*
> 3. A final checkpoint is triggered
> 4. During checkpoint completion, the upstream *CommitterOperator* emits
> *CommittableMessage* instances downstream into the post-commit topology
> 5. The post-commit *SinkWriterOperator.processElement()* rejects these
> elements because *endOfInput* is already *true*
> The *endOfInput* guard assumes no elements arrive after the flag is set, but
> *CommitterOperator* legitimately emits committables during the final
> checkpoint's completion callback, which happens _after_ *endOfInput* has
> propagated.
> This only affects the *SinkWriterOperator* instance _inside the post-commit
> topology_ (the one wrapping the post-commit sink). The primary
> *SinkWriterOperator* (the one receiving user data) is not affected.
> h3. How to reproduce
> The following self-contained program reproduces the bug. It uses a
> *DataGeneratorSource* that emits 100 integers into a sink with a post-commit
> topology that simply discards the committables:
> {code:java}
> import org.apache.flink.api.common.typeinfo.Types;
> import org.apache.flink.api.connector.sink2.Committer;
> import org.apache.flink.api.connector.sink2.CommitterInitContext;
> import org.apache.flink.api.connector.sink2.CommittingSinkWriter;
> import org.apache.flink.api.connector.sink2.Sink;
> import org.apache.flink.api.connector.sink2.SupportsCommitter;
> import org.apache.flink.api.connector.sink2.WriterInitContext;
> import org.apache.flink.connector.datagen.source.DataGeneratorSource;
> import org.apache.flink.core.io.SimpleVersionedSerializer;
> import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
> import
> org.apache.flink.streaming.api.connector.sink2.SupportsPostCommitTopology;
> import org.apache.flink.streaming.api.datastream.DataStream;
> import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
> import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
> import java.io.IOException;
> import java.nio.ByteBuffer;
> import java.util.Collection;
> import java.util.Collections;
> public class PostCommitBoundedSourceBug {
> public static void main(String[] args) throws Exception {
> StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> // Streaming mode is the default, but being explicit:
>
> env.setRuntimeMode(org.apache.flink.api.common.RuntimeExecutionMode.STREAMING);
> env.enableCheckpointing(1000);
> // Bounded source: emits 100 integers, then finishes
> DataGeneratorSource<Integer> source =
> new DataGeneratorSource<>(index -> index.intValue(), 100,
> Types.INT);
> env
> .fromSource(
> source,
>
> org.apache.flink.api.common.eventtime.WatermarkStrategy.noWatermarks(),
> "bounded-source")
> .sinkTo(new PostCommitSink());
> // This will throw:
> // java.lang.IllegalStateException: Received element after endOfInput
> env.execute("post-commit-bounded-source-bug");
> }
> /**
> * Minimal sink with a post-commit topology.
> * The writer produces an Integer committable per element.
> * The committer is a no-op. The post-commit topology discards everything.
> */
> static class PostCommitSink
> implements Sink<Integer>, SupportsCommitter<Integer>,
> SupportsPostCommitTopology<Integer> {
> @Override
> public CommittingSinkWriter<Integer, Integer>
> createWriter(WriterInitContext context) {
> return new CommittingSinkWriter<>() {
> private int pending = 0;
> @Override
> public void write(Integer element, Context context) {
> pending++;
> }
> @Override
> public void flush(boolean endOfInput) {
> }
> @Override
> public Collection<Integer> prepareCommit() {
> int count = pending;
> pending = 0;
> return count > 0 ? Collections.singletonList(count) :
> Collections.emptyList();
> }
> @Override
> public void close() {
> }
> };
> }
> @Override
> public Committer<Integer> createCommitter(CommitterInitContext
> context) {
> return new Committer<Integer>() {
> @Override
> public void commit(Collection<CommitRequest<Integer>>
> committables)
> throws IOException, InterruptedException {
> }
> @Override
> public void close() throws Exception {
> }
> };
> }
> @Override
> public SimpleVersionedSerializer<Integer> getCommittableSerializer() {
> return new SimpleVersionedSerializer<>() {
> @Override
> public int getVersion() {
> return 0;
> }
> @Override
> public byte[] serialize(Integer value) {
> return ByteBuffer.allocate(4).putInt(value).array();
> }
> @Override
> public Integer deserialize(int version, byte[] serialized) {
> return ByteBuffer.wrap(serialized).getInt();
> }
> };
> }
> @Override
> public void
> addPostCommitTopology(DataStream<CommittableMessage<Integer>> committables) {
> // Any .sinkTo() here creates a SinkWriterOperator that will fail
> committables.sinkTo(new DiscardingSink<>());
> }
> }
> }
> {code}
> h3. Expected behavior
> The post-commit topology should successfully process committables emitted
> during the final checkpoint. The job should complete with status
> {*}FINISHED{*}.
> h3. Actual behavior
> The job fails with:
> {code:java}
> java.lang.IllegalStateException: Received element after endOfInput
> at
> org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.processElement(SinkWriterOperator.java:182)
> {code}
> h3. Notes
> This bug does _not_ manifest when:
> - Running in *RuntimeExecutionMode.STREAMING* using an unbounded source (no
> *endOfInput* is ever called)
> - Running in *RuntimeExecutionMode.BATCH*
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
