[
https://issues.apache.org/jira/browse/FLINK-25342?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Hang Ruan updated FLINK-25342:
------------------------------
Attachment: (was: 截屏2021-12-16 下午5.26.31.png)
> 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
>
> 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}
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
