[
https://issues.apache.org/jira/browse/FLINK-25342?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17460531#comment-17460531
]
Fabian Paul commented on FLINK-25342:
-------------------------------------
I think the problem lies in the design of the two sinks. First, the LegacySink
(SinkFunction) is always translated to a single operator therefore we have the
convenience method `addSink`. On the other hand, sinks built with the unified
Sink framework can consist of up to three operators (Writer, Committer,
GlobalCommitter) and the abstraction of holding a single list of operator ids
does not work anymore.
[~ruanhang1993] is this actually a problem or can we close this ticket?
> DataStream.sinkTo will not add sink to the sinks field in the StreamGraph
> -------------------------------------------------------------------------
>
> Key: FLINK-25342
> URL: https://issues.apache.org/jira/browse/FLINK-25342
> Project: Flink
> Issue Type: Bug
> Affects Versions: 1.14.0
> Reporter: Hang Ruan
> Priority: Minor
> Attachments: streamGraph.png
>
>
> I run a test in my IDEA and watch the generated StreamGraph. It seems like
> the sink is not in the field sinks in the StreamGraph. My test is as follows:
> {code:java}
> @Test
> public void selfTest() throws Exception {
> StreamExecutionEnvironment execEnv =
> StreamExecutionEnvironment.getExecutionEnvironment();
> DataStream<String> source = execEnv.fromSource(
> KafkaSource.<String>builder()
> .setGroupId("flink-kafka-test")
> .setDeserializer(
>
> KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class))
> .setTopics("scaleDownTest")
> .setBootstrapServers("localhost:9092")
> .build(),
> WatermarkStrategy.noWatermarks(), "Kafka Source");
> Properties props = new Properties();
> props.setProperty("transaction.timeout.ms", "900000");
> source.sinkTo(KafkaSink.<String>builder()
> .setBootstrapServers("localhost:9092")
> .setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
> .setTransactionalIdPrefix("tp-test-")
> .setKafkaProducerConfig(props)
> .setRecordSerializer(new
> SelfSerializationSchema("scaleDownTestSink1", new SimpleStringSchema()))
> .build());
> execEnv.execute("ScaleDownTest");
> } {code}
> The screen shot lies in the attachments.
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
