flink-1.14.4kafka-2.4.0setTransactionalIdPrefix????????????????????????????????????job??????????????????????????ID????????????????????????????????????chk????????????????????????????ID????????????????????????????????????????????????????????????????System.currentTimeMillis()????????????ID??????????????????????????KafkaSink<String&gt;
 sink = KafkaSink.<String&gt;builder()
        .setBootstrapServers(BOOTSTRAP_SERVERS)
        .setKafkaProducerConfig(kafkaProducerConfig)
        
.setTransactionalIdPrefix(JOB_NAME+System.currentTimeMillis()+"transactional.id")
        .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                .setTopic(SINK_TOPICS)
                .setValueSerializationSchema(new SimpleStringSchema())
                .build())
        .setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
        .build();

回复