kafka 冪等生產者及事務(kafka0.11以後版本新特性)

1. 冪等性設計
1.1 引入目的
生產者重複生產消息。生產者進行retry會產生重試時，會重複產生消息。有了冪等性以後，在進行retry重試時，只會生成一個消息。

1.2 冪等性實現
1.2.1 PID 和 Sequence Number
爲了實現Producer的冪等性，Kafka引入了Producer ID（即PID）和Sequence Number。

PID。每一個新的Producer在初始化的時候會被分配一個惟一的PID，這個PID對用戶是不可見的。
Sequence Numbler。（對於每一個PID，該Producer發送數據的每一個<Topic, Partition>都對應一個從0開始單調遞增的Sequence Number
Broker端在緩存中保存了這seq number，對於接收的每條消息，若是其序號比Broker緩存中序號大於1則接受它，不然將其丟棄。

這樣就能夠實現了消息重複提交了。可是，只能保證單個Producer對於同一個<Topic, Partition>的Exactly Once語義。不能保證同一個Producer一個topic不一樣的partion冪等。

標準實現

 

 

發生重試時

 

 

實現冪等以後

 

 

 發生重試時

 

 

1.2.2  生成PID的流程

//在執行建立事務時
Producer<String, String> producer = new KafkaProducer<String, String>(props);
//會建立一個Sender，並啓動線程，執行以下run方法
Sender{
    void run(long now) {
        if (transactionManager != null) {
            try {
                 ........
                if (!transactionManager.isTransactional()) {
                    // 爲idempotent producer生成一個producer id
                    maybeWaitForProducerId();
                } else if (transactionManager.hasUnresolvedSequences() && !transactionManager.hasFatalError()) {

1.3.演示實例
enable.idempotence，須要設置爲ture,此時就會默認把acks設置爲all，因此不須要再設置acks屬性了。

private Producer buildIdempotProducer(){
         // create instance for properties to access producer configs
        Properties props = new Properties(); 
        // bootstrap.servers是Kafka集羣的IP地址。多個時,使用逗號隔開
        props.put("bootstrap.servers", "localhost:9092"); 
        props.put("enable.idempotence",true); 
        //If the request fails, the producer can automatically retry,
        props.put("retries", 3); 
        //Reduce the no of requests less than 0
        props.put("linger.ms", 1); 
        //The buffer.memory controls the total amount of memory available to the producer for buffering.
        props.put("buffer.memory", 33554432); 
        // Kafka消息是以鍵值對的形式發送,須要設置key和value類型序列化器
        props.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer"); 
        props.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer"); 
        Producer<String, String> producer = new KafkaProducer<String, String>(props);        
        return producer;
}
//發送消息    
public void produceIdempotMessage(String topic, String message) {
        // 建立Producer
        Producer producer = buildIdempotProducer();
        // 發送消息
        producer.send(new ProducerRecord<String, String>(topic, message));
        producer.flush();
}

此時，由於咱們並無配置transaction.id屬性，因此不能使用事務相關API，如：producer.initTransactions();

不然會出現以下錯誤：

Exception in thread 「main」 java.lang.IllegalStateException: Transactional method invoked on a non-transactional producer.

    at org.apache.kafka.clients.producer.internals.TransactionManager.ensureTransactional(TransactionManager.java:777)

    at org.apache.kafka.clients.producer.internals.TransactionManager.initializeTransactions(TransactionManager.java:202)

    at org.apache.kafka.clients.producer.KafkaProducer.initTransactions(KafkaProducer.java:544)

2.事務
2.1 事務屬性
事務屬性是2017年Kafka 0.11.0.0引入的新特性。相似於數據庫事務，只是這裏的數據源是Kafka，kafka事務屬性是指一系列的生產者生產消息和消費者提交偏移量的操做在一個事務，或者說是是一個原子操做），同時成功或者失敗。

注意：在理解消息的事務時，一直處於一個錯誤理解就是以下代碼中，把操做db的業務邏輯跟操做消息當成是一個事務。

其實這個是有問題的，操做DB數據庫的數據源是DB，消息數據源是kfaka，這是徹底不一樣兩個數據，一種數據源（如mysql，kafka）對應一個事務。

因此它們是兩個獨立的事務：kafka事務指kafka一系列 生產、消費消息等操做組成一個原子操做；db事務是指操做數據庫的一系列增刪改操做組成一個原子操做。

void  kakfa_in_tranction(){
  // 1.kafa的操做：讀取消息或者生產消息
 kafkaOperation（）； 
   // 2.db操做
  dbOperation()
 
}

2.2 引入目的
在事務屬性以前先引入了生產者冪等性，它的做用爲：

生產者屢次發送消息能夠封裝成一個原子操做，要麼都成功，要麼失敗
consumer-transform-producer模式下，由於消費者提交偏移量出現問題，致使在重複消費消息時，生產者重複生產消息。

須要將這個模式下消費者提交偏移量操做和生成者一系列生成消息的操做封裝成一個原子操做。
消費者提交偏移量致使重複消費消息的場景：消費者在消費消息完成提交偏移量o2以前掛掉了（假設它最近提交的偏移量是o1），此時執行再均衡時，其它消費者會重複消費消息(o1到o2之間的消息）。

2.3 操做的API

  //producer提供的事務方法
   /**
     * 初始化事務。須要注意的有：
     * 一、前提
     * 須要保證transation.id屬性被配置。
     * 二、這個方法執行邏輯是：
     *   （1）Ensures any transactions initiated by previous instances of the producer with the same
     *      transactional.id are completed. If the previous instance had failed with a transaction in
     *      progress, it will be aborted. If the last transaction had begun completion,
     *      but not yet finished, this method awaits its completion.
     *    （2）Gets the internal producer id and epoch, used in all future transactional
     *      messages issued by the producer.
     *
     */
    public void initTransactions();
 
    /**
     * 開啓事務
     */
    public void beginTransaction() throws ProducerFencedException ;
 
    /**
     * 爲消費者提供的在事務內提交偏移量的操做
     */
    public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                                         String consumerGroupId) throws ProducerFencedException ;
 
    /**
     * 提交事務
     */
    public void commitTransaction() throws ProducerFencedException;
 
    /**
     * 放棄事務，相似回滾事務的操做
     */
    public void abortTransaction() throws ProducerFencedException ;

2.4 演示實例
在一個原子操做中，根據包含的操做類型，能夠分爲三種狀況：

a) 只有Producer生產消息；
b) 消費消息和生產消息並存，這個是事務場景中最經常使用的狀況，就是咱們常說的「consume-transform-produce 」模式
c) 只有consumer消費消息，
前兩種狀況是事務引入的場景，最後一種狀況沒有使用價值（跟使用手動提交效果同樣）。

2.4.1 屬性配置說明
使用kafka的事務api時的一些注意事項：

a) 須要消費者的自動模式設置爲false,而且不能再手動的執行consumer#commitSync或者consumer#commitAsyc
b) 生產者配置transaction.id屬性
c) 生產者不須要再配置enable.idempotence，由於若是配置了transaction.id，則此時enable.idempotence會被設置爲true
d) 消費者須要配置Isolation.level。在consume-trnasform-produce模式下使用事務時，必須設置爲READ_COMMITTED。

2.4.2 只有寫

    /**
     * 在一個事務只有生產消息操做
     */
    public void onlyProduceInTransaction() {
        Producer producer = buildProducer(); 
        // 1.初始化事務
        producer.initTransactions(); 
        // 2.開啓事務
        producer.beginTransaction();
 
        try {
            // 3.kafka寫操做集合
            // 3.1 do業務邏輯 
            // 3.2 發送消息
            producer.send(new ProducerRecord<String, String>("test", "transaction-data-1")); 
            producer.send(new ProducerRecord<String, String>("test", "transaction-data-2"));
            // 3.3 do其餘業務邏輯,還能夠發送其餘topic的消息。
 
            // 4.事務提交
            producer.commitTransaction(); 
 
        } catch (Exception e) {
            // 5.放棄事務
            producer.abortTransaction();
        } 
    }
    /**
     * 須要:
     * 一、設置transactional.id
     * 二、設置enable.idempotence
     * @return
     */
    private Producer buildProducer() { 
        // create instance for properties to access producer configs
        Properties props = new Properties(); 
        // bootstrap.servers是Kafka集羣的IP地址。多個時,使用逗號隔開
        props.put("bootstrap.servers", "localhost:9092"); 
        // 設置事務id
        props.put("transactional.id", "first-transactional"); 
        // 設置冪等性
        props.put("enable.idempotence",true); 
        //Set acknowledgements for producer requests.
        props.put("acks", "all"); 
        //If the request fails, the producer can automatically retry,
        props.put("retries", 1); 
        //Specify buffer size in config,這裏不進行設置這個屬性,若是設置了,還須要執行producer.flush()來把緩存中消息發送出去
        //props.put("batch.size", 16384); 
        //Reduce the no of requests less than 0
        props.put("linger.ms", 1); 
        //The buffer.memory controls the total amount of memory available to the producer for buffering.
        props.put("buffer.memory", 33554432); 
        // Kafka消息是以鍵值對的形式發送,須要設置key和value類型序列化器
        props.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer"); 
        props.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer"); 
        Producer<String, String> producer = new KafkaProducer<String, String>(props); 
        return producer;
    }

2.4.3 消費-生產並存

    /** 
     * 在一個事務內,即有生產消息又有消費消息，即常說的Consume-tansform-produce模式
     */
    public void consumeTransferProduce() {
        // 1.構建上產者
        Producer producer = buildProducer();
        // 2.初始化事務(生成productId),對於一個生產者,只能執行一次初始化事務操做
        producer.initTransactions();
        // 3.構建消費者和訂閱主題
        Consumer consumer = buildConsumer();
        consumer.subscribe(Arrays.asList("test"));
        while (true) {
            // 4.開啓事務
            producer.beginTransaction();
            // 5.1 接受消息
            ConsumerRecords<String, String> records = consumer.poll(500);
            try {
                // 5.2 do業務邏輯;
                System.out.println("customer Message---");
                Map<TopicPartition, OffsetAndMetadata> commits = Maps.newHashMap();
                for (ConsumerRecord<String, String> record : records) {
                    // 5.2.1 讀取消息,並處理消息。print the offset,key and value for the consumer records.
                    System.out.printf("offset = %d, key = %s, value = %s\n",
                            record.offset(), record.key(), record.value());
 
                    // 5.2.2 記錄提交的偏移量
                    commits.put(new TopicPartition(record.topic(), record.partition()),
                            new OffsetAndMetadata(record.offset()));
 
                    // 6.生產新的消息。好比外賣訂單狀態的消息,若是訂單成功,則須要發送跟商家結轉消息或者派送員的提成消息
                    producer.send(new ProducerRecord<String, String>("test", "data2"));
                }
 
                // 7.提交偏移量
                producer.sendOffsetsToTransaction(commits, "group0323");
 
                // 8.事務提交
                producer.commitTransaction();
 
            } catch (Exception e) {
                // 7.放棄事務
                producer.abortTransaction();
            }
        }
    }
    /**
     * 須要:
     * 一、關閉自動提交 enable.auto.commit
     * 二、isolation.level爲read_committed
     * 並且在代碼裏面也不能使用手動提交commitSync( )或者commitAsync( )
     * @return
     */
    public Consumer buildConsumer() {
        Properties props = new Properties();
        // bootstrap.servers是Kafka集羣的IP地址。多個時,使用逗號隔開
        props.put("bootstrap.servers", "localhost:9092");
        // 消費者羣組
        props.put("group.id", "group0323");
        // 設置隔離級別
        props.put("isolation.level","read_committed");
        // 關閉自動提交
        props.put("enable.auto.commit", "false");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer
                <String, String>(props);
        return consumer;
    }

2.4.4 只有讀

    /**
     * 在一個事務只有消費消息操做
     * 這種操做其實沒有什麼意義，跟使用手動提交效果同樣，沒法保證消費消息操做和提交偏移量操做在一個事務。
     */
    public void onlyConsumeInTransaction() {
        Producer producer = buildProducer();
        // 1.初始化事務
        producer.initTransactions();
        // 2.開啓事務
        producer.beginTransaction();
        // 3.kafka讀消息的操做集合
        Consumer consumer = buildConsumer();
        while (true) {
            // 3.1 接受消息
            ConsumerRecords<String, String> records = consumer.poll(500);
 
            try {
                // 3.2 do業務邏輯;
                System.out.println("customer Message---");
                Map<TopicPartition, OffsetAndMetadata> commits = Maps.newHashMap();
                for (ConsumerRecord<String, String> record : records) {
                    // 3.2.1 處理消息 print the offset,key and value for the consumer records.
                    System.out.printf("offset = %d, key = %s, value = %s\n",
                            record.offset(), record.key(), record.value());
 
                    // 3.2.2 記錄提交偏移量
                    commits.put(new TopicPartition(record.topic(), record.partition()),
                            new OffsetAndMetadata(record.offset()));
                }
 
                // 4.提交偏移量
                producer.sendOffsetsToTransaction(commits, "group0323");
 
                // 5.事務提交
                producer.commitTransaction();
 
            } catch (Exception e) {
                // 6.放棄事務
                producer.abortTransaction();
            }
        }
 
    }

3 冪等性和事務性的關係
3.1 二者關係
事務屬性實現前提是冪等性，即在配置事務屬性transaction id時，必須還得配置冪等性；可是冪等性是能夠獨立使用的，不須要依賴事務屬性。

冪等性引入了Porducer ID
事務屬性引入了Transaction Id屬性。
使用場景

enable.idempotence = true，transactional.id不設置：只支持冪等性。
enable.idempotence = true，transactional.id設置：支持事務屬性和冪等性
enable.idempotence = false，transactional.id不設置：沒有事務屬性和冪等性的kafka
enable.idempotence = false，transactional.id設置：沒法獲取到PID，此時會報錯

3.2 tranaction id 、producerId 和 epoch
一個app有一個tid，同一個應用的不一樣實例PID是同樣的，只是epoch的值不一樣。如：

同一份代碼運行兩個實例，分步執行以下：在實例1沒有進行提交事務前，開始執行實例2的初始化事務

step1  實例1-初始化事務。的打印出對應productId和epoch，信息以下：

[2018-04-21 20:56:23,106] INFO [TransactionCoordinator id=0] Initialized transactionalId first-transactional with producerId 8000 and producer epoch 123 on partition __transaction_state-12 (kafka.coordinator.transaction.TransactionCoordinator)

step2 實例1-發送消息。

step3 實例2-初始化事務。初始化事務時的打印出對應productId和epoch，信息以下：

18-04-21 20:56:48,373] INFO [TransactionCoordinator id=0] Initialized transactionalId first-transactional with producerId 8000 and producer epoch 124 on partition __transaction_state-12 (kafka.coordinator.transaction.TransactionCoordinator)

step4  實例1-提交事務，此時報錯

org.apache.kafka.common.errors.ProducerFencedException: Producer attempted an operation with an old epoch. Either there is a newer producer with the same transactionalId, or the producer’s transaction has been expired by the broker.

我今天使用Flink-connector-kafka-0.11時，遇到這個現象

 

step5 實例2-提交事務

爲了不這種錯誤，同一個事務ID，只有保證以下順序epch小producer執行init-transaction和committransaction，而後epoch較大的procuder才能開始執行init-transaction和commit-transaction，以下順序：

有了transactionId後，Kafka可保證：

跨Session的數據冪等發送。當具備相同Transaction ID的新的Producer實例被建立且工做時，舊的且擁有相同Transaction ID的Producer將再也不工做【上面的實例能夠驗證】。

kafka保證了關聯同一個事務的全部producer（一個應用有多個實例）必須按照順序初始化事務、和提交事務，不然就會有問題，這保證了同一事務ID中消息是有序的（不一樣實例得按順序建立事務和提交事務）。

3.3 事務最佳實踐-單實例的事務性
經過上面實例中能夠看到kafka是跨Session的數據冪等發送，即若是應用部署多個實例時常會遇到上面的問題「org.apache.kafka.common.errors.ProducerFencedException: Producer attempted an operation with an old epoch. Either there is a newer producer with the same transactionalId, or the producer’s transaction has been expired by the broker.」，必須保證這些實例生成者的提交事務順序和建立順序保持一致才能夠，不然就沒法成功。其實，在實踐中，咱們更多的是如何實現對應用單實例的事務性。能夠經過spring-kafaka實現思路來學習，即每次建立生成者都設置一個不一樣的transactionId的值，以下代碼：

在spring-kafka中，對於一個線程建立一個producer，事務提交以後，還會關閉這個producer並清除，後續同一個線程或者新的線程從新執行事務時，此時就會從新建立producer。

public class ProducerFactoryUtils{
/**
     * Obtain a Producer that is synchronized with the current transaction, if any.
     * @param producerFactory the ConnectionFactory to obtain a Channel for
     * @param <K> the key type.
     * @param <V> the value type.
     * @return the resource holder.
     */
    public static <K, V> KafkaResourceHolder<K, V> getTransactionalResourceHolder(
            final ProducerFactory<K, V> producerFactory) {
 
        Assert.notNull(producerFactory, "ProducerFactory must not be null");
 
        // 1.對於每個線程會生成一個惟一key，而後根據key去查找resourceHolder
        @SuppressWarnings("unchecked")
        KafkaResourceHolder<K, V> resourceHolder = (KafkaResourceHolder<K, V>) TransactionSynchronizationManager
                .getResource(producerFactory);
        if (resourceHolder == null) {
            // 2.建立一個消費者
            Producer<K, V> producer = producerFactory.createProducer();
            // 3.開啓事務
            producer.beginTransaction();
            resourceHolder = new KafkaResourceHolder<K, V>(producer);
            bindResourceToTransaction(resourceHolder, producerFactory);
        }
        return resourceHolder;
    }
}
//建立消費者代碼
public class DefaultKafkaProducerFactory{
    protected Producer<K, V> createTransactionalProducer() {
        Producer<K, V> producer = this.cache.poll();
        if (producer == null) {
            Map<String, Object> configs = new HashMap<>(this.configs);
            // 對於每一次生成producer時，都設置一個不一樣的transactionId
            configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,
                    this.transactionIdPrefix + this.transactionIdSuffix.getAndIncrement());
            producer = new KafkaProducer<K, V>(configs, this.keySerializer, this.valueSerializer);
            // 1.初始化話事務。
            producer.initTransactions();
            return new CloseSafeProducer<K, V>(producer, this.cache);
        }
        else {
            return producer;
        }
    }
}

3.4 Consume-transform-Produce的流程

 

流程1 ：查找Tranaction Corordinator。

Producer向任意一個brokers發送 FindCoordinatorRequest請求來獲取Transaction Coordinator的地址。

流程2：初始化事務 initTransaction

Producer發送InitpidRequest給事務協調器，獲取一個Pid。InitpidRequest的處理過程是同步阻塞的，一旦該調用正確返回，Producer就能夠開始新的事務。

TranactionalId經過InitpidRequest發送給Tranciton Corordinator，而後在Tranaciton Log中記錄這<TranacionalId,pid>的映射關係。

除了返回PID以外，還具備以下功能：

對PID對應的epoch進行遞增，這樣能夠保證同一個app的不一樣實例對應的PID是同樣的，可是epoch是不一樣的。
回滾以前的Producer未完成的事務（若是有）。
流程3： 開始事務beginTransaction

執行Producer的beginTransacion()，它的做用是Producer在本地記錄下這個transaction的狀態爲開始狀態。

注意：這個操做並無通知Transaction Coordinator。

流程4： Consume-transform-produce loop

流程4.0： 經過Consumtor消費消息，處理業務邏輯

流程4.1： producer向TransactionCordinantro發送AddPartitionsToTxnRequest

在producer執行send操做時，若是是第一次給<topic,partion>發送數據，此時會向Trasaction Corrdinator發送一個AddPartitionsToTxnRequest請求，

Transaction Corrdinator會在transaction log中記錄下tranasactionId和<topic,partion>一個映射關係，並將狀態改成begin。AddPartionsToTxnRequest的數據結構以下：

      AddPartitionsToTxnRequest => TransactionalId PID Epoch [Topic [Partition]]
      TransactionalId => string
      PID => int64
      Epoch => int16
      Topic => string
      Partition => int32

流程4.2：  producer#send發送 ProduceRequst，生產者發送數據，雖然沒有尚未執行commit或者absrot，可是此時消息已經保存到kafka上，

能夠參考以下圖斷點位置處，此時已經能夠查看到消息了，並且即便後面執行abort，消息也不會刪除，只是更改狀態字段標識消息爲abort狀態。

 

流程4.3： AddOffsetCommitsToTxnRequest，Producer經過KafkaProducer.sendOffsetsToTransaction 向事務協調器器發送一個AddOffesetCommitsToTxnRequests：

    AddOffsetsToTxnRequest => TransactionalId PID Epoch ConsumerGroupID
    TransactionalId => string
     PID => int64
     Epoch => int16
     ConsumerGroupID => string

在執行事務提交時，能夠根據ConsumerGroupID來推斷_customer_offsets主題中相應的TopicPartions信息。這樣在

流程4.4: TxnOffsetCommitRequest，Producer經過KafkaProducer.sendOffsetsToTransaction還會向消費者協調器Cosumer Corrdinator發送一個TxnOffsetCommitRequest，在主題_consumer_offsets中保存消費者的偏移量信息。

TxnOffsetCommitRequest   => ConsumerGroupID
                            PID
                            Epoch
                            RetentionTime
                            OffsetAndMetadata
  ConsumerGroupID => string
  PID => int64
  Epoch => int32
  RetentionTime => int64
  OffsetAndMetadata => [TopicName [Partition Offset Metadata]]
    TopicName => string
    Partition => int32
    Offset => int64
    Metadata => string

流程5： 事務提交和事務終結(放棄事務)，經過生產者的commitTransaction或abortTransaction方法來提交事務和終結事務，這兩個操做都會發送一個EndTxnRequest給Transaction Coordinator。

流程5.1：EndTxnRequest。Producer發送一個EndTxnRequest給Transaction Coordinator，而後執行以下操做：

Transaction Coordinator會把PREPARE_COMMIT or PREPARE_ABORT 消息寫入到transaction log中記錄
執行流程5.2
執行流程5.3
流程5.2：WriteTxnMarkerRequest

WriteTxnMarkersRequest => [CoorinadorEpoch PID Epoch Marker [Topic [Partition]]]
 CoordinatorEpoch => int32
 PID => int64
 Epoch => int16
 Marker => boolean (false(0) means ABORT, true(1) means COMMIT)
 Topic => string
 Partition => int32

對於Producer生產的消息。Tranaction Coordinator會發送WriteTxnMarkerRequest給當前事務涉及到每一個<topic,partion>的leader，leader收到請求後，會寫入一個COMMIT(PID) 或者 ABORT(PID)的控制信息到data log中
對於消費者偏移量信息，若是在這個事務裏面包含_consumer-offsets主題。Tranaction Coordinator會發送WriteTxnMarkerRequest給Transaction Coordinartor，Transaction Coordinartor收到請求後，

會寫入一個COMMIT(PID) 或者 ABORT(PID)的控制信息到 data log中
流程5.3：Transaction Coordinator會將最終的COMPLETE_COMMIT或COMPLETE_ABORT消息寫入Transaction Log中以標明該事務結束。

只會保留這個事務對應的PID和timstamp。而後把當前事務其餘相關消息刪除掉，包括PID和tranactionId的映射關係。

3.4.1 文件類型和查看命令

 kafka文件主要包括broker的data（主題：test）、事務協調器對應的transaction_log（主題：__tranaction_state）、偏移量信息（主題:_consumer_offsets）三種類型。

以下圖

這三種文件類型其實都是topic的分區，因此對於每個目錄都包含*.log、*.index、*.timeindex、*.txnindex文件（僅這個文件是爲了實現事務屬性引入的）。

查看文件內容：

bin/kafka-run-class.sh kafka.tools.DumpLogSegments   –files /kafka-logs/firtstopic-0/00000000000000000002.log   –print-data-log