简单介绍Kafka Topic

Kafka的topic可以看作一个partition的queue, 通过发给topic时指定partition(或者用一个partitioner 比如按key做hash来指定使用那个partition), 不同的key的消息可以发送到不同的partition, 相同key的message则可以保证发送到同一个partition去, topic里的信息可以靠一个确定的index来访问, 就好像一个数据库一样，所以只要在data retention到期之前，consumer都可以用同一个index来访问之前已经访问过的数据；

Kafka Transactional Messaging

前边说过, Kafka Stream是建立在kafka分布式队列功能上个一个library, 主要依靠kafka的Transactional Messaging来实现end2end exactly once msg processing;

Transactional Messaging是指用户可以通过类似以下code来定义哪些对kafka topic的写属于一个transaction, 并进一步保证tx的atomic和Isolation

producer.initTransactions();
 try { 
    // called  right before sending any records 
    producer.beginTransaction();
    //sending some messages...
    // when done  sending, commit the transaction 
    producer.commitTransaction();
} catch  (Exception e) {
     producer.close();
} catch  (KafkaException e) {
    producer.abortTransaction();  
} 

Kafka transaction保证了, beginTransactions之后的, 所有往不同Kafka topic里发送的消息, 要么在commitTransaction之后才能被read-committed的consumer看到, 要么由于close或者failure而全部作废, 从而不为read-committed的consumer所见;

而kafka stream通过用kafka本身的分布式queue的功能来实现了state和记录处理event进度的功能; 因为

(1) 所有的要发送的计算结果(由于可以允许计算发不同消息给多个下游,所以可能发给不同的topic和partition) (2)记录input event stream的消费进度, (3)所有的state的所有更新, 这3点, Kafka Stream都是用写消息到kafka topic里实现的, (1)自不必说，本来就是往topic里写数据，(2)其实是写当前consume position的topic; (注意Kafka Stream的上下游消息传递考的是一个中间隐藏的Kafka topic, 上游往这个topic写, 下游从这个topic读, 上游不需要下游的ack，只要往topic里写成功即可, 也不需要管下游已经处理到那里了；而下游则需要维护自己"处理到那里了"这个信息，储存在consume position的topic, 这样比如机器挂掉需要在另外的host上重启计算节点，则计算节点可以从记录consume position的topic里读出自己处理到那里然后从失败的地方重洗开始) (3)其实是写一个内部隐藏的state的change log的topic,和一个本地key value表(也就是本计算节点的state); failover的时候, 之前的"本地"表丢失没关系, 可以冲change log里恢复出失败前确定commit的所有state;

(1)(2)(3)的topic都只是普通的Kafka topic; 只不过(2)(3)由Kafka Stream自动创建和维护(一部分用来支持高层API的(1)也是自动创建)

• 开始计算时, 在从上游的topic里拿msg之前, Kafka Stream会启动一个tx, 然后开始才开始计算, 此时tx coordinator会分配一个新的epoch id给这个producer并且以后跟tx coordinator通讯都要附带这个epochId
• Kafka Stream的计算节点的上游信息都来自kafka topic的分布式partition queue, 且只接收commit之后的record, 在queue里的record都有确定的某种sequenceId, 所以只要计算节点记录好自己当前处理的sequenceId, 处理完一个信息就更新自己的sequenceId到下一个record, 且commit到可靠dataStore里, 就绝对不会重复处理上游event, 而只要没有commit这个位置则可以无数次replay当前的record; (b.下游去重, c. 要求触发本次计算的event可以replay)
• 在tx内部，每从上游topic里读一条信息就写一条信息到记录consume position的topic里, 每一个state的更改都会更新到本地的state(是一张表)里，且同时写在隐藏的changelog里; 计算过程中需要往下游发信息则写与下游联系的topic；
• 计算结束后, commit本次的tx, 由Kafka Transactional Messaging来保证本次tx里发生的所有(1)往下游发的消息, (2) 记录input event stream的消费进度,(3)所有的state的所有更新是一个原子操作, 由于结果都成功写入kafka topic，所以达到计算结果的高可用性 (a.要求结果高可用, d+.要求记录event处理的进度, 并保证储存计算结果和state的更新不出现重复)
• 计算过程中出现failure(比如机器挂了), 那么当计算重启，会重新运行initTransactions来注册tx, 此时tx coordinator会分配一个新的epoch id给此producer, 并且从此以后拒绝老的epoch id的任何commit信息来防止zombie的计算节点; tx coordinator同时roll back(如果上一个tx已经在prepare_commit状态, 继续完成transaction, 具体看下边Transactional Messaging这个章节); 如果rollback，那么input的处理进度, 状态的更改和往下游发送的信息都会rollback, 那么计算可以重新开始，就好像没有上次fail的失败一样; 如果上一个tx已经prepare_commit, 那么完成所有信息的commit; 此时当initTransactions返回，当前计算会接着上一个tx完成的进度继续计算;(e. state需要在replay event的时候rollback到处理event之前时的状态)