Kafka 幾個重要的配置總結

注意：配置基於Kafka 0.8.2.1

broker配置

#非負整數，用於惟一標識broker
broker.id 0

#kafka持久化數據存儲的路徑，能夠指定多個，以逗號分隔
log.dirs /tmp/kafka-logs

#broker接收鏈接請求的端口
port 9092

#指定zk鏈接字符串，[hostname:port]以逗號分隔
zookeeper.connect

#單條消息最大大小控制，消費端的最大拉取大小須要略大於該值
message.max.bytes 1000000

#接收網絡請求的線程數
num.network.threads 3

#用於執行請求的I/O線程數
num.io.threads 8

#用於各類後臺處理任務（如文件刪除）的線程數
background.threads 10

#待處理請求最大可緩衝的隊列大小
queued.max.requests 500

#配置該機器的IP地址
host.name 

#默認分區個數
num.partitions  1

#分段文件大小，超事後會輪轉
log.segment.bytes 1024 * 1024 * 1024

#日誌沒達到大小，若是達到這個時間也會輪轉
log.roll.{ms,hours}   168

#日誌保留時間
log.retention.{ms,minutes,hours}

#不存在topic的時候是否自動建立
auto.create.topics.enable true

#partition默認的備份因子
default.replication.factor 1

#若是這個時間內follower沒有發起fetch請求，被認爲dead，從ISR移除
replica.lag.time.max.ms   10000

#若是follower相比leader落後這麼多以上消息條數，會被從ISR移除
replica.lag.max.messages  4000

#從leader能夠拉取的消息最大大小
replica.fetch.max.bytes 1024 * 1024

#從leader拉取消息的fetch線程數
num.replica.fetchers 1

#zk會話超時時間
zookeeper.session.timeout.ms  6000

#zk鏈接所用時間
zookeeper.connection.timeout.ms

#zk follower落後leader的時間
zookeeper.sync.time.ms 2000

#是否開啓topic能夠被刪除的方式
delete.topic.enable false

producer配置

#參與消息確認的broker數量控制，0表明不須要任何確認 1表明須要leader replica確認 -1表明須要ISR中全部進行確認
request.required.acks 0

#從發送請求到收到ACK確認等待的最長時間（超時時間）
request.timeout.ms  10000

#設置消息發送模式，默認是同步方式， async異步模式下容許消息累計到必定量或一段時間又另外線程批量發送，吞吐量好但丟失數據風險增大
producer.type sync

#消息序列化類實現方式，默認是byte[]數組形式
serializer.class kafka.serializer.DefaultEncoder

#kafka消息分區策略實現方式，默認是對key進行hash
partitioner.class kafka.producer.DefaultPartitioner

#對發送的消息採起的壓縮編碼方式，有none|gzip|snappy
compression.codec none

#指定哪些topic的message須要壓縮
compressed.topics  null

#消息發送失敗的狀況下，重試發送的次數 存在消息發送是成功的，只是因爲網絡致使ACK沒收到的重試，會出現消息被重複發送的狀況
message.send.max.retries 3

#在開始從新發起metadata更新操做須要等待的時間
retry.backoff.ms 100

#metadata刷新間隔時間，若是負值則失敗的時候纔會刷新，若是0則每次發送後都刷新，正值則是一種週期行爲
topic.metadata.refresh.interval.ms 600 * 1000

#異步發送模式下，緩存數據的最長時間，以後便會被髮送到broker
queue.buffering.max.ms 5000

#producer端異步模式下最多緩存的消息條數
queue.buffering.max.messages 10000

#0表明隊列沒滿的時候直接入隊，滿了當即扔棄，-1表明無條件阻塞且不丟棄
queue.enqueue.timeout.ms -1

#一次批量發送須要達到的消息條數，固然若是queue.buffering.max.ms達到的時候也會被髮送
batch.num.messages 200

consumer配置

#指明當前消費進程所屬的消費組，一個partition只能被同一個消費組的一個消費者消費
group.id

#針對一個partition的fetch request所能拉取的最大消息字節數，必須大於等於Kafka運行的最大消息
fetch.message.max.bytes  1024 * 1024

#是否自動週期性提交已經拉取到消費端的消息offset
auto.commit.enable true

#自動提交offset到zookeeper的時間間隔
auto.commit.interval.ms  60 * 1000

#消費均衡的重試次數
rebalance.max.retries  4

#消費均衡兩次重試之間的時間間隔
rebalance.backoff.ms 2000

#當從新去獲取partition的leader前須要等待的時間
refresh.leader.backoff.ms   200

#若是zookeeper上沒有offset合理的初始值狀況下獲取第一條消息開始的策略smallest|largeset
auto.offset.reset largest

#若是其超時，將會可能觸發rebalance並認爲已經死去
zookeeper.session.timeout.ms  6000

#確認zookeeper鏈接創建操做客戶端能等待的最長時間
zookeeper.connection.timeout.ms 6000

1.maven:

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka_2.10</artifactId>
  <version>0.8.2.1</version>
</dependency>
2.kafka生產者代碼：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
 
import java.util.Properties;
 
/**
 * 
 * @author FromX
 *
 */
public class KProducer {
 
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        //kafka服務器地址
        props.put("bootstrap.servers", "slave1.com:6667,slave2.com:6667,slave3.com:6667");
        //ack是判斷請求是否爲完整的條件（即判斷是否成功發送）。all將會阻塞消息，這種設置性能最低，可是最可靠。
        props.put("acks", "1");
        //retries,若是請求失敗，生產者會自動重試，咱們指定是0次，若是啓用重試，則會有重複消息的可能性。
        props.put("retries", 0);
        //producer緩存每一個分區未發送消息，緩存的大小是經過batch.size()配置設定的。值較大的話將會產生更大的批。並須要更多的內存(由於每一個「活躍」的分區都有一個緩衝區)
        props.put("batch.size", 16384);
        //默認緩衝區可當即發送，即使緩衝區空間沒有滿；可是，若是你想減小請求的數量，能夠設置linger.ms大於0.這將指示生產者發送請求以前等待一段時間
        //但願更多的消息補填到未滿的批中。這相似於tcp的算法，例如上面的代碼段，可能100條消息在一個請求發送，由於咱們設置了linger時間爲1ms，而後，若是咱們
        //沒有填滿緩衝區，這個設置將增長1ms的延遲請求以等待更多的消息。須要注意的是，在高負載下，相近的時間通常也會組成批，即便是linger.ms=0。
        //不處於高負載的狀況下，若是設置比0大，以少許的延遲代價換取更少的，更有效的請求。
        props.put("linger.ms", 1);
        //buffer.memory控制生產者可用的緩存總量，若是消息發送速度比其傳輸到服務器的快，將會耗盡這個緩存空間。當緩存空間耗盡，其餘發送調用將被阻塞，阻塞時間的閾值
        //經過max.block.ms設定，以後他將拋出一個TimeoutExecption。
        props.put("buffer.memory", 33554432);
        //key.serializer和value.serializer示例：將用戶提供的key和value對象ProducerRecord轉換成字節，你可使用附帶的ByteArraySerizlizaer或StringSerializer處理簡單的byte和String類型.
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //設置kafka的分區數量
        props.put("kafka.partitions", 12);
        
        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 50; i++){
            System.out.println("key-->key"+i+"  value-->vvv"+i);
            producer.send(new ProducerRecord<String, String>("aaa", "key"+i, "vvv"+i));
            Thread.sleep(1000);
        }
              
        producer.close();
    }
}
3.kafka消費者代碼：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
 
import java.util.Arrays;
import java.util.Properties;
 
/**
 * 
 * @author FromX
 *
 */
public class KConsumer {
 
    public KafkaConsumer<String, String>  getConsmer() {
        Properties props = new Properties();
        //設置kafka服務器
        props.put("bootstrap.servers", "c1.wb3.com:6667,n1.wb1.com:6667");
        //消費者羣組ID，發佈-訂閱模式，即若是一個生產者，多個消費者都要消費，那麼須要定義本身的羣組，同一個羣組內的消費者只有一個能消費到消息
        props.put("group.id", "test");
        //true，消費者的偏移量將在後臺按期提交；false關閉自動提交位移，在消息被完整處理以後再手動提交位移
        props.put("enable.auto.commit", "true");
        //如何設置爲自動提交（enable.auto.commit=true）,這裏設置自動提交週期
        props.put("auto.commit.interval.ms", "1000");
        //session.timeout.ms:在使用kafka的組管理時，用於檢測消費者故障的超時
        props.put("session.timeout.ms", "30000");
        //key.serializer和value.serializer示例：將用戶提供的key和value對象ProducerRecord轉換成字節，你可使用附帶的ByteArraySerizlizaer或StringSerializer處理簡單的byte和String類型.
        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<>(props);
        return consumer;
    }
 
    
    
    public static void main(String[] args) {
        KConsumer kconsumer =  new KConsumer();
        KafkaConsumer<String, String> consumer = kconsumer.getConsmer();
        
        consumer.subscribe(Arrays.asList("aaa"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
                System.out.println("offset =  "+record.offset()+", key = "+record.key()+", value = "+ record.value());
        }
    }
}
4.官網文檔地址：http://kafka.apache.org/documentation.html#configuration

5.極限狀況的數據丟失現象

a:即便將ack設置爲"all"也會在必定狀況下丟失消息，由於kafka的高性能特性，消息在寫入kafka時並無落盤而是寫入了OS buffer中，使用Os的髒頁刷新策略週期性落盤，就算落盤 仍然會有raid buffer。前者機器宕機數據丟失，後者機器跳電數據丟失。

b:對數據可靠性比較高的場景建議offset手動提交，自動提交當遇到業務系統上線關閉時，消息讀取而且offset已經提交，可是數據沒有儲存或者仍沒來得及消費時，消息狀態在內存中沒法保留，重啓應用會跳過消息，導致消息丟失。
--------------------- 

Kafka設計解析（八）- Kafka事務機制與Exactly Once語義實現原理

未分類

---

---

description: 本文介紹了Kafka實現事務性的幾個階段——正好一次語義與原子操做。以後詳細分析了Kafka事務機制的實現原理，並介紹了Kafka如何處理事務相關的異常狀況，如Transaction Coordinator宕機。最後介紹了Kafka的事務機制與PostgreSQL的MVCC以及Zookeeper的原子廣播實現事務的異同

---

寫在前面的話

本文全部Kafka原理性的描述除特殊說明外均基於Kafka 1.0.0版本。

爲何要提供事務機制

Kafka事務機制的實現主要是爲了支持

• Exactly Once即正好一次語義
• 操做的原子性
• 有狀態操做的可恢復性

Exactly Once

《Kafka背景及架構介紹》一文中有說明Kafka在0.11.0.0以前的版本中只支持At Least Once和At Most Once語義，尚不支持Exactly Once語義。

可是在不少要求嚴格的場景下，如使用Kafka處理交易數據，Exactly Once語義是必須的。咱們能夠經過讓下游系統具備冪等性來配合Kafka的At Least Once語義來間接實現Exactly Once。可是：

• 該方案要求下游系統支持冪等操做，限制了Kafka的適用場景
• 實現門檻相對較高，須要用戶對Kafka的工做機制很是瞭解
• 對於Kafka Stream而言，Kafka自己便是本身的下游系統，但Kafka在0.11.0.0版本以前不具備冪等發送能力

所以，Kafka自己對Exactly Once語義的支持就很是必要。

操做原子性

操做的原子性是指，多個操做要麼所有成功要麼所有失敗，不存在部分紅功部分失敗的可能。

實現原子性操做的意義在於：

• 操做結果更可控，有助於提高數據一致性
• 便於故障恢復。由於操做是原子的，從故障中恢復時只須要重試該操做（若是原操做失敗）或者直接跳過該操做（若是原操做成功），而不須要記錄中間狀態，更不須要針對中間狀態做特殊處理

實現事務機制的幾個階段

冪等性發送

上文提到，實現Exactly Once的一種方法是讓下游系統具備冪等處理特性，而在Kafka Stream中，Kafka Producer自己就是「下游」系統，所以若是能讓Producer具備冪等處理特性，那就可讓Kafka Stream在必定程度上支持Exactly once語義。

爲了實現Producer的冪等語義，Kafka引入了Producer ID（即PID）和Sequence Number。每一個新的Producer在初始化的時候會被分配一個惟一的PID，該PID對用戶徹底透明而不會暴露給用戶。

對於每一個PID，該Producer發送數據的每一個<Topic, Partition>都對應一個從0開始單調遞增的Sequence Number。

相似地，Broker端也會爲每一個<PID, Topic, Partition>維護一個序號，而且每次Commit一條消息時將其對應序號遞增。對於接收的每條消息，若是其序號比Broker維護的序號（即最後一次Commit的消息的序號）大一，則Broker會接受它，不然將其丟棄：

• 若是消息序號比Broker維護的序號大一以上，說明中間有數據還沒有寫入，也即亂序，此時Broker拒絕該消息，Producer拋出InvalidSequenceNumber
• 若是消息序號小於等於Broker維護的序號，說明該消息已被保存，即爲重複消息，Broker直接丟棄該消息，Producer拋出DuplicateSequenceNumber

上述設計解決了0.11.0.0以前版本中的兩個問題：

• Broker保存消息後，發送ACK前宕機，Producer認爲消息未發送成功並重試，形成數據重複
• 前一條消息發送失敗，後一條消息發送成功，前一條消息重試後成功，形成數據亂序

事務性保證

上述冪等設計只能保證單個Producer對於同一個<Topic, Partition>的Exactly Once語義。

另外，它並不能保證寫操做的原子性——即多個寫操做，要麼所有被Commit要麼所有不被Commit。

更不能保證多個讀寫操做的的原子性。尤爲對於Kafka Stream應用而言，典型的操做便是從某個Topic消費數據，通過一系列轉換後寫回另外一個Topic，保證從源Topic的讀取與向目標Topic的寫入的原子性有助於從故障中恢復。

事務保證可以使得應用程序將生產數據和消費數據看成一個原子單元來處理，要麼所有成功，要麼所有失敗，即便該生產或消費跨多個<Topic, Partition>。

另外，有狀態的應用也能夠保證重啓後從斷點處繼續處理，也即事務恢復。

爲了實現這種效果，應用程序必須提供一個穩定的（重啓後不變）惟一的ID，也即Transaction ID。Transactin ID與PID可能一一對應。區別在於Transaction ID由用戶提供，而PID是內部的實現對用戶透明。

另外，爲了保證新的Producer啓動後，舊的具備相同Transaction ID的Producer即失效，每次Producer經過Transaction ID拿到PID的同時，還會獲取一個單調遞增的epoch。因爲舊的Producer的epoch比新Producer的epoch小，Kafka能夠很容易識別出該Producer是老的Producer並拒絕其請求。

有了Transaction ID後，Kafka可保證：

• 跨Session的數據冪等發送。當具備相同Transaction ID的新的Producer實例被建立且工做時，舊的且擁有相同Transaction ID的Producer將再也不工做。
• 跨Session的事務恢復。若是某個應用實例宕機，新的實例能夠保證任何未完成的舊的事務要麼Commit要麼Abort，使得新實例從一個正常狀態開始工做。

須要注意的是，上述的事務保證是從Producer的角度去考慮的。從Consumer的角度來看，該保證會相對弱一些。尤爲是不能保證全部被某事務Commit過的全部消息都被一塊兒消費，由於：

• 對於壓縮的Topic而言，同一事務的某些消息可能被其它版本覆蓋
• 事務包含的消息可能分佈在多個Segment中（即便在同一個Partition內），當老的Segment被刪除時，該事務的部分數據可能會丟失
• Consumer在一個事務內可能經過seek方法訪問任意Offset的消息，從而可能丟失部分消息
• Consumer可能並不須要消費某一事務內的全部Partition，所以它將永遠不會讀取組成該事務的全部消息

事務機制原理

事務性消息傳遞

這一節所說的事務主要指原子性，也即Producer將多條消息做爲一個事務批量發送，要麼所有成功要麼所有失敗。

爲了實現這一點，Kafka 0.11.0.0引入了一個服務器端的模塊，名爲Transaction Coordinator，用於管理Producer發送的消息的事務性。

該Transaction Coordinator維護Transaction Log，該log存於一個內部的Topic內。因爲Topic數據具備持久性，所以事務的狀態也具備持久性。

Producer並不直接讀寫Transaction Log，它與Transaction Coordinator通訊，而後由Transaction Coordinator將該事務的狀態插入相應的Transaction Log。

Transaction Log的設計與Offset Log用於保存Consumer的Offset相似。

事務中Offset的提交

許多基於Kafka的應用，尤爲是Kafka Stream應用中同時包含Consumer和Producer，前者負責從Kafka中獲取消息，後者負責將處理完的數據寫回Kafka的其它Topic中。

爲了實現該場景下的事務的原子性，Kafka須要保證對Consumer Offset的Commit與Producer對發送消息的Commit包含在同一個事務中。不然，若是在兩者Commit中間發生異常，根據兩者Commit的順序可能會形成數據丟失和數據重複：

• 若是先Commit Producer發送數據的事務再Commit Consumer的Offset，即At Least Once語義，可能形成數據重複。
• 若是先Commit Consumer的Offset，再Commit Producer數據發送事務，即At Most Once語義，可能形成數據丟失。

用於事務特性的控制型消息

爲了區分寫入Partition的消息被Commit仍是Abort，Kafka引入了一種特殊類型的消息，即Control Message。該類消息的Value內不包含任何應用相關的數據，而且不會暴露給應用程序。它只用於Broker與Client間的內部通訊。

對於Producer端事務，Kafka以Control Message的形式引入一系列的Transaction Marker。Consumer便可經過該標記斷定對應的消息被Commit了仍是Abort了，而後結合該Consumer配置的隔離級別決定是否應該將該消息返回給應用程序。

事務處理樣例代碼

1. Producer<String, String> producer = new KafkaProducer<String, String>(props);
2.  
3. // 初始化事務，包括結束該Transaction ID對應的未完成的事務（若是有）
4. // 保證新的事務在一個正確的狀態下啓動
5. producer.initTransactions();
6.  
7. // 開始事務
8. producer.beginTransaction();
9.  
10. // 消費數據
11. ConsumerRecords<String, String> records = consumer.poll(100);
12.  
13. try{
14. // 發送數據
15. producer.send(new ProducerRecord<String, String>("Topic", "Key", "Value"));
16.  
17. // 發送消費數據的Offset，將上述數據消費與數據發送歸入同一個Transaction內
18. producer.sendOffsetsToTransaction(offsets, "group1");
19.  
20. // 數據發送及Offset發送均成功的狀況下，提交事務
21. producer.commitTransaction();
22. } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
23. // 數據發送或者Offset發送出現異常時，終止事務
24. producer.abortTransaction();
25. } finally {
26. // 關閉Producer和Consumer
27. producer.close();
28. consumer.close();
29. }

完整事務過程

找到Transaction Coordinator

因爲Transaction Coordinator是分配PID和管理事務的核心，所以Producer要作的第一件事情就是經過向任意一個Broker發送FindCoordinator請求找到Transaction Coordinator的位置。

注意：只有應用程序爲Producer配置了Transaction ID時纔可以使用事務特性，也才須要這一步。另外，因爲事務性要求Producer開啓冪等特性，所以經過將transactional.id設置爲非空從而開啓事務特性的同時也須要經過將enable.idempotence設置爲true來開啓冪等特性。

獲取PID

找到Transaction Coordinator後，具備冪等特性的Producer必須發起InitPidRequest請求以獲取PID。

注意：只要開啓了冪等特性即必須執行該操做，而無須考慮該Producer是否開啓了事務特性。

* 若是事務特性被開啓 * 
InitPidRequest會發送給Transaction Coordinator。若是Transaction Coordinator是第一次收到包含有該Transaction ID的InitPidRequest請求，它將會把該<TransactionID, PID>存入Transaction Log，如上圖中步驟2.1所示。這樣可保證該對應關係被持久化，從而保證即便Transaction Coordinator宕機該對應關係也不會丟失。

除了返回PID外，InitPidRequest還會執行以下任務：

• 增長該PID對應的epoch。具備相同PID但epoch小於該epoch的其它Producer（若是有）新開啓的事務將被拒絕。
• 恢復（Commit或Abort）以前的Producer未完成的事務（若是有）。

注意：InitPidRequest的處理過程是同步阻塞的。一旦該調用正確返回，Producer便可開始新的事務。

另外，若是事務特性未開啓，InitPidRequest可發送至任意Broker，而且會獲得一個全新的惟一的PID。該Producer將只能使用冪等特性以及單一Session內的事務特性，而不能使用跨Session的事務特性。

開啓事務

Kafka從0.11.0.0版本開始，提供beginTransaction()方法用於開啓一個事務。調用該方法後，Producer本地會記錄已經開啓了事務，但Transaction Coordinator只有在Producer發送第一條消息後才認爲事務已經開啓。

Consume-Transform-Produce

這一階段，包含了整個事務的數據處理過程，而且包含了多種請求。

AddPartitionsToTxnRequest 
一個Producer可能會給多個<Topic, Partition>發送數據，給一個新的<Topic, Partition>發送數據前，它須要先向Transaction Coordinator發送AddPartitionsToTxnRequest。

Transaction Coordinator會將該<Transaction, Topic, Partition>存於Transaction Log內，並將其狀態置爲BEGIN，如上圖中步驟4.1所示。有了該信息後，咱們才能夠在後續步驟中爲每一個Topic, Partition>設置COMMIT或者ABORT標記（如上圖中步驟5.2所示）。

另外，若是該<Topic, Partition>爲該事務中第一個<Topic, Partition>，Transaction Coordinator還會啓動對該事務的計時（每一個事務都有本身的超時時間）。

ProduceRequest 
Producer經過一個或多個ProduceRequest發送一系列消息。除了應用數據外，該請求還包含了PID，epoch，和Sequence Number。該過程如上圖中步驟4.2所示。

AddOffsetsToTxnRequest 
爲了提供事務性，Producer新增了sendOffsetsToTransaction方法，該方法將多組消息的發送和消費放入同一批處理內。

該方法先判斷在當前事務中該方法是否已經被調用並傳入了相同的Group ID。如果，直接跳到下一步；若不是，則向Transaction Coordinator發送AddOffsetsToTxnRequests請求，Transaction Coordinator將對應的全部<Topic, Partition>存於Transaction Log中，並將其狀態記爲BEGIN，如上圖中步驟4.3所示。該方法會阻塞直到收到響應。

TxnOffsetCommitRequest 
做爲sendOffsetsToTransaction方法的一部分，在處理完AddOffsetsToTxnRequest後，Producer也會發送TxnOffsetCommit請求給Consumer Coordinator從而將本事務包含的與讀操做相關的各<Topic, Partition>的Offset持久化到內部的__consumer_offsets中，如上圖步驟4.4所示。

在此過程當中，Consumer Coordinator會經過PID和對應的epoch來驗證是否應該容許該Producer的該請求。

這裏須要注意：

• 寫入__consumer_offsets的Offset信息在當前事務Commit前對外是不可見的。也即在當前事務被Commit前，可認爲該Offset還沒有Commit，也即對應的消息還沒有被完成處理。
• Consumer Coordinator並不會當即更新緩存中相應<Topic, Partition>的Offset，由於此時這些更新操做還沒有被COMMIT或ABORT。

Commit或Abort事務

一旦上述數據寫入操做完成，應用程序必須調用KafkaProducer的commitTransaction方法或者abortTransaction方法以結束當前事務。

EndTxnRequest 
commitTransaction方法使得Producer寫入的數據對下游Consumer可見。abortTransaction方法經過Transaction Marker將Producer寫入的數據標記爲Aborted狀態。下游的Consumer若是將isolation.level設置爲READ_COMMITTED，則它讀到被Abort的消息後直接將其丟棄而不會返回給客戶程序，也即被Abort的消息對應用程序不可見。

不管是Commit仍是Abort，Producer都會發送EndTxnRequest請求給Transaction Coordinator，並經過標誌位標識是應該Commit仍是Abort。

收到該請求後，Transaction Coordinator會進行以下操做

1. 將PREPARE_COMMIT或PREPARE_ABORT消息寫入Transaction Log，如上圖中步驟5.1所示
2. 經過WriteTxnMarker請求以Transaction Marker的形式將COMMIT或ABORT信息寫入用戶數據日誌以及Offset Log中，如上圖中步驟5.2所示
3. 最後將COMPLETE_COMMIT或COMPLETE_ABORT信息寫入Transaction Log中，如上圖中步驟5.3所示

補充說明：對於commitTransaction方法，它會在發送EndTxnRequest以前先調用flush方法以確保全部發送出去的數據都獲得相應的ACK。對於abortTransaction方法，在發送EndTxnRequest以前直接將當前Buffer中的事務性消息（若是有）所有丟棄，但必須等待全部被髮送但還沒有收到ACK的消息發送完成。

上述第二步是實現將一組讀操做與寫操做做爲一個事務處理的關鍵。由於Producer寫入的數據Topic以及記錄Comsumer Offset的Topic會被寫入相同的Transactin Marker，因此這一組讀操做與寫操做要麼所有COMMIT要麼所有ABORT。

WriteTxnMarkerRequest 
上面提到的WriteTxnMarkerRequest由Transaction Coordinator發送給當前事務涉及到的每一個<Topic, Partition>的Leader。收到該請求後，對應的Leader會將對應的COMMIT(PID)或者ABORT(PID)控制信息寫入日誌，如上圖中步驟5.2所示。

該控制消息向Broker以及Consumer代表對應PID的消息被Commit了仍是被Abort了。

這裏要注意，若是事務也涉及到__consumer_offsets，即該事務中有消費數據的操做且將該消費的Offset存於__consumer_offsets中，Transaction Coordinator也須要向該內部Topic的各Partition的Leader發送WriteTxnMarkerRequest從而寫入COMMIT(PID)或COMMIT(PID)控制信息。

寫入最終的COMPLETE_COMMIT或COMPLETE_ABORT消息 
寫完全部的Transaction Marker後，Transaction Coordinator會將最終的COMPLETE_COMMIT或COMPLETE_ABORT消息寫入Transaction Log中以標明該事務結束，如上圖中步驟5.3所示。

此時，Transaction Log中全部關於該事務的消息所有能夠移除。固然，因爲Kafka內數據是Append Only的，不可直接更新和刪除，這裏說的移除只是將其標記爲null從而在Log Compact時再也不保留。

另外，COMPLETE_COMMIT或COMPLETE_ABORT的寫入並不須要獲得全部Rreplica的ACK，由於若是該消息丟失，能夠根據事務協議重發。

補充說明，若是參與該事務的某些<Topic, Partition>在被寫入Transaction Marker前不可用，它對READ_COMMITTED的Consumer不可見，但不影響其它可用<Topic, Partition>的COMMIT或ABORT。在該<Topic, Partition>恢復可用後，Transaction Coordinator會從新根據PREPARE_COMMIT或PREPARE_ABORT向該<Topic, Partition>發送Transaction Marker。

總結

• PID與Sequence Number的引入實現了寫操做的冪等性
• 寫操做的冪等性結合At Least Once語義實現了單一Session內的Exactly Once語義
• Transaction Marker與PID提供了識別消息是否應該被讀取的能力，從而實現了事務的隔離性
• Offset的更新標記了消息是否被讀取，從而將對讀操做的事務處理轉換成了對寫（Offset）操做的事務處理
• Kafka事務的本質是，將一組寫操做（若是有）對應的消息與一組讀操做（若是有）對應的Offset的更新進行一樣的標記（即Transaction Marker）來實現事務中涉及的全部讀寫操做同時對外可見或同時對外不可見
• Kafka只提供對Kafka自己的讀寫操做的事務性，不提供包含外部系統的事務性

異常處理

Exception處理

InvalidProducerEpoch 
這是一種Fatal Error，它說明當前Producer是一個過時的實例，有Transaction ID相同但epoch更新的Producer實例被建立並使用。此時Producer會中止並拋出Exception。

InvalidPidMapping 
Transaction Coordinator沒有與該Transaction ID對應的PID。此時Producer會經過包含有Transaction ID的InitPidRequest請求建立一個新的PID。

NotCorrdinatorForGTransactionalId 
該Transaction Coordinator不負責該當前事務。Producer會經過FindCoordinatorRequest請求從新尋找對應的Transaction Coordinator。

InvalidTxnRequest 
違反了事務協議。正確的Client實現不該該出現這種Exception。若是該異常發生了，用戶須要檢查本身的客戶端實現是否有問題。

CoordinatorNotAvailable 
Transaction Coordinator仍在初始化中。Producer只須要重試便可。

DuplicateSequenceNumber 
發送的消息的序號低於Broker預期。該異常說明該消息已經被成功處理過，Producer能夠直接忽略該異常並處理下一條消息

InvalidSequenceNumber 
這是一個Fatal Error，它說明發送的消息中的序號大於Broker預期。此時有兩種可能

• 數據亂序。好比前面的消息發送失敗後重試期間，新的消息被接收。正常狀況下不該該出現該問題，由於當冪等發送啓用時，max.inflight.requests.per.connection被強制設置爲1，而acks被強制設置爲all。故前面消息重試期間，後續消息不會被髮送，也即不會發生亂序。而且只有ISR中全部Replica都ACK，Producer纔會認爲消息已經被髮送，也即不存在Broker端數據丟失問題。
• 服務器因爲日誌被Truncate而形成數據丟失。此時應該中止Producer並將此Fatal Error報告給用戶。

InvalidTransactionTimeout 
InitPidRequest調用出現的Fatal Error。它代表Producer傳入的timeout時間不在可接受範圍內，應該中止Producer並報告給用戶。

處理Transaction Coordinator失敗

寫PREPARE_COMMIT/PREPARE_ABORT前失敗

Producer經過FindCoordinatorRequest找到新的Transaction Coordinator，並經過EndTxnRequest請求發起COMMIT或ABORT流程，新的Transaction Coordinator繼續處理EndTxnRequest請求——寫PREPARE_COMMIT或PREPARE_ABORT，寫Transaction Marker，寫COMPLETE_COMMIT或COMPLETE_ABORT。

寫完PREPARE_COMMIT/PREPARE_ABORT後失敗

此時舊的Transaction Coordinator可能已經成功寫入部分Transaction Marker。新的Transaction Coordinator會重複這些操做，因此部分Partition中可能會存在重複的COMMIT或ABORT，但只要該Producer在此期間沒有發起新的事務，這些重複的Transaction Marker就不是問題。

寫完COMPLETE_COMMIT/ABORT後失敗

舊的Transaction Coordinator可能已經寫完了COMPLETE_COMMIT或COMPLETE_ABORT但在返回EndTxnRequest以前失敗。該場景下，新的Transaction Coordinator會直接給Producer返回成功。

事務過時機制

事務超時

transaction.timeout.ms

終止過時事務

當Producer失敗時，Transaction Coordinator必須可以主動的讓某些進行中的事務過時。不然沒有Producer的參與，Transaction Coordinator沒法判斷這些事務應該如何處理，這會形成：

• 若是這種進行中事務太多，會形成Transaction Coordinator須要維護大量的事務狀態，大量佔用內存
• Transaction Log內也會存在大量數據，形成新的Transaction Coordinator啓動緩慢
• READ_COMMITTED的Consumer須要緩存大量的消息，形成沒必要要的內存浪費甚至是OOM
• 若是多個Transaction ID不一樣的Producer交叉寫同一個Partition，當一個Producer的事務狀態不更新時，READ_COMMITTED的Consumer爲了保證順序消費而被阻塞

爲了不上述問題，Transaction Coordinator會週期性遍歷內存中的事務狀態Map，並執行以下操做

• 若是狀態是BEGIN而且其最後更新時間與當前時間差大於transaction.remove.expired.transaction.cleanup.interval.ms（默認值爲1小時），則主動將其終止：1）未避免原Producer臨時恢復與當前終止流程衝突，增長該Producer對應的PID的epoch，並確保將該更新的信息寫入Transaction Log；2）以更新後的epoch回滾事務，從而使得該事務相關的全部Broker都更新其緩存的該PID的epoch從而拒絕舊Producer的寫操做
• 若是狀態是PREPARE_COMMIT，完成後續的COMMIT流程————向各<Topic, Partition>寫入Transaction Marker，在Transaction Log內寫入COMPLETE_COMMIT
• 若是狀態是PREPARE_ABORT，完成後續ABORT流程

終止Transaction ID

某Transaction ID的Producer可能很長時間再也不發送數據，Transaction Coordinator不必再保存該Transaction ID與PID等的映射，不然可能會形成大量的資源浪費。所以須要有一個機制探測再也不活躍的Transaction ID並將其信息刪除。

Transaction Coordinator會週期性遍歷內存中的Transaction ID與PID映射，若是某Transaction ID沒有對應的正在進行中的事務而且它對應的最後一個事務的結束時間與當前時間差大於transactional.id.expiration.ms（默認值是7天），則將其從內存中刪除並在Transaction Log中將其對應的日誌的值設置爲null從而使得Log Compact可將其記錄刪除。

與其它系統事務機制對比

PostgreSQL MVCC

Kafka的事務機制與《MVCC PostgreSQL實現事務和多版本併發控制的精華》一文中介紹的PostgreSQL經過MVCC實現事務的機制很是相似，對於事務的回滾，並不須要刪除已寫入的數據，都是將寫入數據的事務標記爲Rollback/Abort從而在讀數據時過濾該數據。

兩階段提交

Kafka的事務機制與《分佈式事務（一）兩階段提交及JTA》一文中所介紹的兩階段提交機制看似類似，都分PREPARE階段和最終COMMIT階段，但又有很大不一樣。

• Kafka事務機制中，PREPARE時即要指明是PREPARE_COMMIT仍是PREPARE_ABORT，而且只須在Transaction Log中標記便可，無須其它組件參與。而兩階段提交的PREPARE須要發送給全部的分佈式事務參與方，而且事務參與方須要儘量準備好，並根據準備狀況返回Prepared或Non-Prepared狀態給事務管理器。
• Kafka事務中，一但發起PREPARE_COMMIT或PREPARE_ABORT，則肯定該事務最終的結果應該是被COMMIT或ABORT。而分佈式事務中，PREPARE後由各事務參與方返回狀態，只有全部參與方均返回Prepared狀態纔會真正執行COMMIT，不然執行ROLLBACK
• Kafka事務機制中，某幾個Partition在COMMIT或ABORT過程當中變爲不可用，隻影響該Partition不影響其它Partition。兩階段提交中，若惟一收到COMMIT命令參與者Crash，其它事務參與方沒法判斷事務狀態從而使得整個事務阻塞
• Kafka事務機制引入事務超時機制，有效避免了掛起的事務影響其它事務的問題
• Kafka事務機制中存在多個Transaction Coordinator實例，而分佈式事務中只有一個事務管理器

Zookeeper

Zookeeper的原子廣播協議與兩階段提交以及Kafka事務機制有類似之處，但又有各自的特色

• Kafka事務可COMMIT也可ABORT。而Zookeeper原子廣播協議只有COMMIT沒有ABORT。固然，Zookeeper不COMMIT某消息也即等效於ABORT該消息的更新。
• Kafka存在多個Transaction Coordinator實例，擴展性較好。而Zookeeper寫操做只能在Leader節點進行，因此其寫性能遠低於讀性能。
• Kafka事務是COMMIT仍是ABORT徹底取決於Producer即客戶端。而Zookeeper原子廣播協議中某條消息是否被COMMIT取決因而否有一大半FOLLOWER ACK該消息。

Kafka系列文章

• Kafka設計解析（一）- Kafka背景及架構介紹
• Kafka設計解析（二）- Kafka High Availability （上）
• Kafka設計解析（三）- Kafka High Availability （下）
• Kafka設計解析（四）- Kafka Consumer設計解析
• Kafka設計解析（五）- Kafka性能測試方法及Benchmark報告
• Kafka設計解析（六）- Kafka高性能架構之道
• Kafka設計解析（七）- Kafka Stream