lvyanquan commented on code in PR #25331:
URL: https://github.com/apache/flink/pull/25331#discussion_r1778847740
##########
flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java:
##########
@@ -85,35 +81,39 @@ public static void main(String[] args) throws Exception {
SingleOutputStreamOperator<String> mapToString =
input.map((MapFunction<User, String>)
SpecificRecordBase::toString);
- KafkaSink<String> stringSink =
- KafkaSink.<String>builder()
- .setBootstrapServers(bootstrapServers)
- .setRecordSerializer(
- KafkaRecordSerializationSchema.builder()
- .setValueSerializationSchema(new
SimpleStringSchema())
-
.setTopic(parameterTool.getRequired("output-string-topic"))
- .build())
- .setKafkaProducerConfig(config)
- .build();
- mapToString.sinkTo(stringSink);
-
- KafkaSink<User> avroSink =
- KafkaSink.<User>builder()
- .setBootstrapServers(bootstrapServers)
- .setRecordSerializer(
- KafkaRecordSerializationSchema.builder()
- .setValueSerializationSchema(
-
ConfluentRegistryAvroSerializationSchema
- .forSpecific(
- User.class,
-
parameterTool.getRequired(
-
"output-subject"),
-
schemaRegistryUrl))
-
.setTopic(parameterTool.getRequired("output-avro-topic"))
- .build())
- .build();
- input.sinkTo(avroSink);
-
- env.execute("Kafka Confluent Schema Registry AVRO Example");
+ // TODO: [FLINK-36245] Release comments after KafkaSink does not rely
on the Depreciated API
Review Comment:
Actually, KafkaSink implements StatefulSink/TwoPhaseCommittingSink, these
two APIs were removed in current version and will be replaced by
SupportsWriterState/SupportsCommitter.
If this code is kept, it will result in unsuccessful compilation.
--
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]
