FLINK -KAFKA连接器恰好一次错误

Question

我正在使用Flink 1.15.0。 For Kafka integration I wrote

KafkaSource:

public static KafkaSource<String> kafkaSource(String bootstrapServers, String topic, String groupId) {
    return KafkaSource.<String>builder()
            .setBootstrapServers(bootstrapServers)
            .setTopics(topic)              
            .setGroupId(groupId)                
            .setStartingOffsets(OffsetsInitializer.latest())
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();

and KafkaSink:

public static KafkaSink<WordCountPojo> kafkaSink(String brokers, String topic, Properties producerProperties) {
    return KafkaSink.<WordCountPojo>builder()
            .setKafkaProducerConfig(producerProperties)
            .setBootstrapServers(brokers)
            .setRecordSerializer(new WordCountPojoSerializer(topic))
            .setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
            .setTransactionalIdPrefix(UUID.randomUUID().toString())
            .build();
}

for completeness this is my custom serializer

public class WordCountPojoSerializer implements KafkaRecordSerializationSchema<WordCountPojo> {
    private String topic;
    private ObjectMapper mapper;

    public WordCountPojoSerializer(String topic) {
        this.topic = topic;
        mapper = new ObjectMapper();
    }

    @Override
    public ProducerRecord<byte[], byte[]> serialize(WordCountPojo wordCountPojo, KafkaSinkContext kafkaSinkContext, Long timestamp) {
        try {
            byte[] serializedValue = mapper.writeValueAsBytes(wordCountPojo);
            return new ProducerRecord<>(topic, null,  timestamp,null, serializedValue);
        } catch (JsonProcessingException e) {
            return null;
        }
    }
}

Property transaction.timeout.ms is set to 60_000ms (1 minute)

My application run for a while, then, suddenly, stop除了

Caused by: org.apache.kafka.common.errors.InvalidProducerEpochException: Producer attempted to produce with an old epoch.

生产者似乎试图在两项交易之间产生记录，但我不确定。 我可以将Transaction.mms增加到最大900_000（15分钟），但我认为这可能无法解决问题。

你能帮我吗？