机制查看源码发现是使用两阶段提交策略而事务提交是checkpoint完成后才执行那么如果checkpoint设置间隔时间比较长时事务未提交之前后端应该消费不到数据而观察实际现象为写入kafka的消费数据可以立马消费。

测试用例

CheckpointingMode.EXACTLY_ONCE输出输出到kafka。

编写任务2消费任务的结果topic打印控制台验证结果。

根据现象查看源码分析原因。

测试用例

StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);env.enableCheckpointing(1000*60l,

CheckpointingMode.EXACTLY_ONCE);env.getCheckpointConfig().setCheckpointStorage(file:///flink/checkpoint);//

超时时间checkpoint没在时间内完成则丢弃env.getCheckpointConfig().setCheckpointTimeout(50000L);

//10秒env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);env.getCheckpointConfig().setTolerableCheckpointFailureNumber(1);//最小间隔时间前一次结束时间与下一次开始时间间隔env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);

Flink

KafkaSource.Stringbuilder().setBootstrapServers(127.0.0.1:9092).setTopics(test001).setGroupId(my-group)

.setStartingOffsets(OffsetsInitializer()).setStartingOffsets(OffsetsInitializer.committedOffsets()).setValueOnlyDeserializer(new

SimpleStringSchema()).build();DataStreamSourceString

kafkaSource

WatermarkStrategy.noWatermarks(),

Kafka

根据上面的介绍自己计算这边的超时时间满足条件即可properties.setProperty(transaction.timeout.ms,900000);

properties.setProperty(bootstrap.servers,

127.0.0.1:9092);KafkaSinkString

sink

KafkaSink.Stringbuilder().setBootstrapServers(192.168.65.128:9092).setRecordSerializer(KafkaRecordSerializationSchema.Stringbuilder().setTopic(test002).setValueSerializationSchema(new

SimpleStringSchema()).build()).setKafkaProducerConfig(properties).setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)

.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE).setTransactionalIdPrefix(flink-xhaodream-).build();map.sinkTo(sink);//

打印输出env.execute();

StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);env.enableCheckpointing(1000*150l,

CheckpointingMode.EXACTLY_ONCE);env.getCheckpointConfig().setCheckpointStorage(file:///flink/checkpoint);

Flink

properties1.put(isolation.level,read_committed);KafkaSourceString

source

KafkaSource.Stringbuilder().setBootstrapServers(127.0.0.1:9092).setTopics(test002).setGroupId(my-group2).setProperties(properties1).setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST)).setValueOnlyDeserializer(new

SimpleStringSchema()).build();DataStreamSourceString

kafkaSource

WatermarkStrategy.noWatermarks(),

Kafka

任务1开启后无论是否执行checkpoint任务checkpoint都可以正常消费数据与预期不符合。

原因排查

的源码找到跟与两阶段提交相关的代码1.18源码中TwoPhaseCommittingSink有重构。

kafkasink实现TwoPhaseCommittingSink接口实现创建Commiter和Writer。

PublicEvolving

{PrecommittingSinkWriterInputT,

CommT

IOException;SimpleVersionedSerializerCommT

getCommittableSerializer();PublicEvolvingpublic

interface

}--------------------------------------

public

KafkaWriterState,TwoPhaseCommittingSinkIN,

KafkaCommittable

KafkaRecordSerializationSchemaIN

final

transactionalIdPrefix;KafkaSink(DeliveryGuarantee

transactionalIdPrefix,KafkaRecordSerializationSchemaIN

recordSerializer)

deliveryGuarantee;this.kafkaProducerConfig

kafkaProducerConfig;this.transactionalIdPrefix

transactionalIdPrefix;this.recordSerializer

Create

创建CommitterInternalOverridepublic

createCommitter()

KafkaCommitter(kafkaProducerConfig);}InternalOverridepublic

SimpleVersionedSerializerKafkaCommittable

{return

KafkaWriterIN(deliveryGuarantee,kafkaProducerConfig,transactionalIdPrefix,context,recordSerializer,context.asSerializationSchemaInitializationContext(),Collections.emptyList());}InternalOverridepublic

KafkaWriterIN

KafkaWriter(deliveryGuarantee,kafkaProducerConfig,transactionalIdPrefix,context,recordSerializer,context.asSerializationSchemaInitializationContext(),recoveredState);}InternalOverridepublic

SimpleVersionedSerializerKafkaWriterState

{return

KafkaWriterStateSerializer();}VisibleForTestingprotected

Properties

在KafkaWriter中snapshotState方法中发现如果deliveryGuarantee

DeliveryGuarantee.EXACTLY_ONCE的开启事务的判断逻辑。

class

StatefulSink.StatefulSinkWriterIN,

KafkaWriterState,TwoPhaseCommittingSink.PrecommittingSinkWriterIN,

KafkaCommittable

DeliveryGuarantee.EXACTLY_ONCE)

{return

(currentProducer.hasRecordsInTransaction())

{final

Collections.singletonList(KafkaCommittable.of(currentProducer,

producerPool::add));LOG.debug(Committing

committables.,

producercurrentProducer.commitTransaction();producerPool.add(currentProducer);return

Collections.emptyList();}Overridepublic

ListKafkaWriterState

DeliveryGuarantee.EXACTLY_ONCE)

{currentProducer

getTransactionalProducer(checkpointId

1);currentProducer.beginTransaction();}return

Collections.singletonList(kafkaWriterState);}

KafkaCommitter的commit()方法发现producer.commitTransaction();操作

/***

LoggerFactory.getLogger(KafkaCommitter.class);public

static

UNKNOWN_PRODUCER_ID_ERROR_MESSAGE

because

recoveryProducer;KafkaCommitter(Properties

kafkaProducerConfig)

kafkaProducerConfig;}Overridepublic

void

commit(CollectionCommitRequestKafkaCommittable

requests)throws

committable.getTransactionalId();LOG.debug(Committing

Kafka

transactionalId);OptionalRecyclable?

extends

committable.getProducer();FlinkKafkaInternalProducer?,

producer;try

recyclable.FlinkKafkaInternalProducer?,

?map(Recyclable::getObject).orElseGet(()

getRecoveryProducer(committable));---

事务提交producer.commitTransaction();producer.flush();recyclable.ifPresent(Recyclable::close);}

catch

发现除了设置checkpoint还需要kafkasink单独设置.才会实现输出端的开启事务因此在任务1中添加设置setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)

KafkaSinkString

KafkaSink.Stringbuilder().setBootstrapServers(192.168.65.128:9092).setRecordSerializer(KafkaRecordSerializationSchema.Stringbuilder().setTopic(test002).setValueSerializationSchema(new

SimpleStringSchema()).build()).setKafkaProducerConfig(properties).setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)

.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE).setTransactionalIdPrefix(flink-xhaodream-).build();

再次验证任务任务2依然可以正常消费。

这是有一点头大不明白为什么想到既然开启事务肯定有事务的隔离级别查询了kafka的事务隔离级别有两种分别是读已提交和读未提交默认消费事务是读未提交。

读已提交Read

committed此隔离级别保证消费者只能读取已经提交的消息。

这意味着事务中的消息在提交之前对消费者是不可见的。

使用此隔离级别可以避免消费者读取到未提交的事务消息确保消费者只读取到已经持久化的消息。

读未提交Read

Uncommitted此隔离级别允许消费者读取未提交的消息。

这意味着事务中的消息在提交之前就对消费者可见。

使用此隔离级别可以实现更低的延迟但可能会导致消费者读取到未提交的事务消息。

​

在任务2中添加isolation.levelread_committed设定读取消费事务级别为读已提交再次测试发现任务1执行完checkpoint前任务2消费不到数据。

而命令行可以及时消费任务1的输出topic可可以消费到数据。

结果与预期相同。

Properties

Properties();properties1.put(isolation.level,read_committed);KafkaSourceString

source

KafkaSource.Stringbuilder().setBootstrapServers(127.0.0.1:9092).setTopics(test002).setGroupId(my-group2).setProperties(properties1)

注意事项

setCommitOffsetsOnCheckpoints(boolean)

方法来禁用或启用

总共支持三种不同的语义保证DeliveryGuarantee。

对于

DeliveryGuarantee.AT_LEAST_ONCE

DeliveryGuarantee.EXACTLY_ONCEFlink

checkpoint

的故障发生重复。

DeliveryGuarantee.AT_LEAST_ONCE:

sink

会重新处理旧数据。

DeliveryGuarantee.EXACTLY_ONCE:

该模式下Kafka

的前缀transactionIdPrefix对不同的应用是唯一的以保证不同作业的事务

不会互相影响此外强烈建议将

推荐查看1.14版本和1.18版本结合起来看在一些细节处理上有差异。

Kafka

如果查看1.18版本源码不太好理解两阶段提交可以查看1.14.5的源码发现FlinkKafkaProducer被标记废除请改用

KafkaSink并将在

在1.14.5中TwoPhaseCommitSinkFunction为抽象类有明确定开启事务、预提交和提交的抽象方法比较好理解。

的依赖发现没有直接使用TwoPhaseCommitSinkFunction但是查看源码可以看到使用了commiter和kafkawriter对象

public

context.metricGroup().addGroup(user);return

new

KafkaWriter(deliveryGuarantee,kafkaProducerConfig,transactionalIdPrefix,context,recordSerializer,new

InitContextInitializationContextAdapter(context.getUserCodeClassLoader(),

metricGroupSupplier),states);}--

事务提交在kafkaCommitterOverridepublic

OptionalCommitterKafkaCommittable

createCommitter()

KafkaCommitter(kafkaProducerConfig));}Overridepublic

OptionalGlobalCommitterKafkaCommittable,

Void

DeliveryGuarantee.EXACTLY_ONCE)

{final

Collections.singletonList(KafkaCommittable.of(currentProducer,

producerPool::add));LOG.debug(Committing

committables,

DeliveryGuarantee.EXACTLY_ONCE)

{currentProducer

getTransactionalProducer(checkpointId

1);currentProducer.beginTransaction();}return

ImmutableList.of(kafkaWriterState);}1.14.5

版本TwoPhaseCommitSinkFunction是一个抽象类

在1.18

FlinkKafkaProducer.Semantic#AT_LEAST_ONCE}

semantic.

FlinkKafkaProducer.Semantic#EXACTLY_ONCE}

please

org.apache.flink.connector.kafka.sink.KafkaSink}.*/

Deprecated

TwoPhaseCommitSinkFunctionIN,FlinkKafkaProducer.KafkaTransactionState,FlinkKafkaProducer.KafkaTransactionContext

1.14

{PrecommittingSinkWriterInputT,

CommT

IOException;SimpleVersionedSerializerCommT

getCommittableSerializer();PublicEvolvingpublic

interface

FlinkKafkaProducer继承TwoPhaseCommitSinkFunction会重写其中的方法查看重写开启事务的方法--

FlinkKafkaProducer

FlinkKafkaProducer.KafkaTransactionState

FlinkKafkaException

EXACTLY_ONCE:FlinkKafkaInternalProducerbyte[],

byte[]

开启kafka的procder的事务producer.beginTransaction();return

new

FlinkKafkaProducer.KafkaTransactionState(producer.getTransactionalId(),

producer);case

FlinkKafkaProducer.KafkaTransactionState

currentTransaction

FlinkKafkaProducer.KafkaTransactionState(currentTransaction.producer);}return

new

FlinkKafkaProducer.KafkaTransactionState(initNonTransactionalProducer(true));default:throw

new

UnsupportedOperationException(Not

implemented

customPartitioner,FlinkKafkaProducer.Semantic

semantic,int

KafkaSerializationSchemaWrapper(topicId,

customPartitioner,

serializationSchema),producerConfig,semantic,kafkaProducersPoolSize);}