0.11版本引入 幂等 和 事务


幂等Producer

引入ProducerID和SequenceNumber

ProducerID
每个新的Producer初始化时，分配一个唯一的ProducerID，对客户端使用者不可见

SequenceNumber
对于每个ProducerID，Producer发送数据的每个Topic和Partition都对应一个从0开始单调递增的SequenceNumber值


props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true)。
public static final String ENABLE_IDEMPOTENCE_CONFIG = "enable.idempotence";

自动去重，Broker会多保存一些字段

只能保证单分区上的幂等性
只能实现会话幂等性，重启Producer之后，幂等性不保证


read committed隔离级别事务
保证多条消息原子性地写入到目标分区，同时也保证Consumer只能看到事务成功提交的消息


Producer端设置

enable.idempotence = true
设置参数 transactional.id

调用事务API
initTransaction
beginTransaction
commitTransaction
abortTransaction

producer.initTransactions();
try {
            producer.beginTransaction();
            producer.send(record1);
            producer.send(record2);
            producer.commitTransaction();
} catch (KafkaException e) {
            producer.abortTransaction();
}

保证record1和record2被当做一个事务提交到Kafka，要么全部成功，要么全部失败。


Consumer端设置
设置isolation.level参数 

read_uncommitted
默认值  事务型Producer提交事务还是终止事务，写入的消息都可以读取

read_committed
只会读取事务型Producer成功提交事务写入的消息。
非事务型Producer写入的所有消息都能看到。