捡贝壳: kafka精确一次投递和事务消息学习整理

概述

今年6月发布的kafka 0.11.0.0包含两个比较大的特性，exactly once delivery和transactional transactional messaging。之前一直对事务这块比较感兴趣，所以抽空详细学习了一下，感觉收获还是挺多的。

对这两个特性的详细描述可以看这三篇文档，

https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging

https://cwiki.apache.org/confluence/display/KAFKA/Idempotent+Producer

https://cwiki.apache.org/confluence/display/KAFKA/Transactional+Messaging+in+Kafka

需求场景

精确一次投递

消息重复一直是消息领域的一个痛点，而消息重复可能发生于下面这些场景

1.消息发送端发出消息，服务端落盘以后因为网络等种种原因发送端得到一个发送失败的响应，然后发送端重发消息导致消息重复。

2.消息消费端在消费过程中挂掉另一个消费端启动拿之前记录的位点开始消费，由于位点的滞后性可能会导致新启动的客户端有少量重复消费。

先说说问题2，一般的解决方案是让下游做幂等或者尽量每消费一条消息都记位点，对于少数严格的场景可能需要把位点和下游状态更新放在同一个数据库里面做事务来保证精确的一次更新或者在下游数据表里面同时记录消费位点，然后更新下游数据的时候用消费位点做乐观锁拒绝掉旧位点的数据更新。

问题1的解决方案也就是kafka实现的方案是每个producer有一个producer id，服务端会通过这个id关联记录每个producer的状态，每个producer的每条消息会带上一个递增的sequence，服务端会记录每个producer对应的当前最大sequence，如果新的消息带上的sequence不大于当前的最大sequence就拒绝这条消息，问题1的场景如果消息落盘会同时更新最大sequence，这个时候重发的消息会被服务端拒掉从而避免消息重复。后面展开详细说一下这个解决方案。

事务消息

1.最简单的需求是producer发的多条消息组成一个事务这些消息需要对consumer同时可见或者同时不可见
2.producer可能会给多个topic，多个partition发消息，这些消息也需要能放在一个事务里面，这就形成了一个典型的分布式事务
3.kafka的应用场景经常是应用先消费一个topic，然后做处理再发到另一个topic，这个consume-transform-produce过程需要放到一个事务里面，比如在消息处理或者发送的过程中如果失败了，消费位点也不能提交
4.producer或者producer所在的应用可能会挂掉，新的producer启动以后需要知道怎么处理之前未完成的事务
5.流式处理的拓扑可能会比较深，如果下游只有等上游消息事务提交以后才能读到，可能会导致rt非常长吞吐量也随之下降很多，所以需要实现read committed和read uncommitted两种事务隔离级别

一个比较典型的consume-transform-produce的场景像下面这样

public class KafkaTransactionsExample {
  
  public static void main(String args[]) {
    KafkaConsumer consumer = new KafkaConsumer<>(consumerConfig);
    KafkaProducer producer = new KafkaProducer<>(producerConfig);

    producer.initTransactions();
     
    while(true) {
      ConsumerRecords records = consumer.poll(CONSUMER_POLL_TIMEOUT);
      if (!records.isEmpty()) {
        producer.beginTransaction();
        List> outputRecords = processRecords(records);
        for (ProducerRecord outputRecord : outputRecords) {
          producer.send(outputRecord);
        }
        sendOffsetsResult = producer.sendOffsetsToTransaction(getUncommittedOffsets());
        producer.endTransaction();
      }
    }
  }
}

几个关键概念和推导

1.因为producer发送消息可能是分布式事务，所以引入了常用的2PC，所以有事务协调者(Transaction Coordinator)。Transaction Coordinator和之前为了解决脑裂和惊群问题引入的Group Coordinator在选举和failover上面类似。

2.事务管理中事务日志是必不可少的，kafka使用一个内部topic来保存事务日志，这个设计和之前使用内部topic保存位点的设计保持一致。事务日志是Transaction Coordinator管理的状态的持久化，因为不需要回溯事务的历史状态，所以事务日志只用保存最近的事务状态。
3.因为事务存在commit和abort两种操作，而客户端又有read committed和read uncommitted两种隔离级别，所以消息队列必须能标识事务状态，这个被称作Control Message。
4.producer挂掉重启或者漂移到其它机器需要能关联的之前的未完成事务所以需要有一个唯一标识符来进行关联，这个就是TransactionalId，一个producer挂了，另一个有相同TransactionalId的producer能够接着处理这个事务未完成的状态。注意不要把TransactionalId和数据库事务中常见的transaction id搞混了，kafka目前没有引入全局序，所以也没有transaction id，这个TransactionalId是用户提前配置的。
5. TransactionalId能关联producer，也需要避免两个使用相同TransactionalId的producer同时存在，所以引入了producer epoch来保证对应一个TransactionalId只有一个活跃的producer epoch

重要的类图

架构和数据流

事务消息

官方文档的数据流组件图

上图中每个方框代表一台独立的机器，图下方比较长的圆角矩形代表kafka topic，图中间的两个角是圆的的方框代表broker里面的逻辑组件，箭头代表rpc调用。

接下来说一下事务的数据流，这里基本按照官方文档的结构加上我自己看代码的一点补充

1.首先producer需要找到transaction coordinator。TransactionManager.lookupCoordinator

private synchronized void lookupCoordinator(FindCoordinatorRequest.CoordinatorType type, String coordinatorKey) {
    switch (type) {
        case GROUP:
            consumerGroupCoordinator = null;
            break;
        case TRANSACTION:
            transactionCoordinator = null;
            break;
        default:
            throw new IllegalStateException("Invalid coordinator type: " + type);
    }

    FindCoordinatorRequest.Builder builder = new FindCoordinatorRequest.Builder(type, coordinatorKey);
    enqueueRequest(new FindCoordinatorHandler(builder));
}

2.获取producer id，producer id是比较重要的概念，精确一次投递需要producer id+sequence防止重复投递，事务消息也需要保存transactional id和producer id的对应关系。客户端调用KafkaProducer.initTransactions的时候会向coordinator请求producer id，TransactionManager.initializeTransactions

public synchronized TransactionalRequestResult initializeTransactions() {
    ensureTransactional();
    transitionTo(State.INITIALIZING);
    setProducerIdAndEpoch(ProducerIdAndEpoch.NONE);
    this.sequenceNumbers.clear();
    InitProducerIdRequest.Builder builder = new InitProducerIdRequest.Builder(transactionalId, transactionTimeoutMs);
    InitProducerIdHandler handler = new InitProducerIdHandler(builder);
    enqueueRequest(handler);
    return handler.result;
}

coordinator端处理逻辑在TransactionCoordinator.handleInitProducerId流程比较复杂，首先如果对应的transactional id没有产生过producer id会找producerIdManager生成一个

val coordinatorEpochAndMetadata = txnManager.getTransactionState(transactionalId).right.flatMap {
  case None =>
    val producerId = producerIdManager.generateProducerId()
    val createdMetadata = new TransactionMetadata(transactionalId = transactionalId,
      producerId = producerId,
      producerEpoch = RecordBatch.NO_PRODUCER_EPOCH,
      txnTimeoutMs = transactionTimeoutMs,
      state = Empty,
      topicPartitions = collection.mutable.Set.empty[TopicPartition],
      txnLastUpdateTimestamp = time.milliseconds())
    txnManager.putTransactionStateIfNotExists(transactionalId, createdMetadata)

  case Some(epochAndTxnMetadata) => Right(epochAndTxnMetadata)
}

producer id需要全局唯一，有点类似于tddl sequence的生成逻辑，ProducerIdManager.generateProducerId会一次申请一批id然后在zk上面保存状态，本地每次生成+1，如果超出了当前批次的范围就去找zk重新申请

拿到了producer id接下来处理事务状态，保证之前的事务状态能够处理完毕，该提交的提交，该回滚的回滚。
TransactionCoordinator.prepareInitProduceIdTransit处理producer id的变化比如开始一个新的事务可能会增加producer epoch，也可能生成新的producer id

case PrepareAbort | PrepareCommit =>
  // reply to client and let it backoff and retry
  Left(Errors.CONCURRENT_TRANSACTIONS)

case CompleteAbort | CompleteCommit | Empty =>
  val transitMetadata = if (txnMetadata.isProducerEpochExhausted) {
    val newProducerId = producerIdManager.generateProducerId()
    txnMetadata.prepareProducerIdRotation(newProducerId, transactionTimeoutMs, time.milliseconds())
  } else {
    txnMetadata.prepareIncrementProducerEpoch(transactionTimeoutMs, time.milliseconds())
  }

  Right(coordinatorEpoch, transitMetadata)

case Ongoing =>
  // indicate to abort the current ongoing txn first. Note that this epoch is never returned to the
  // user. We will abort the ongoing transaction and return CONCURRENT_TRANSACTIONS to the client.
  // This forces the client to retry, which will ensure that the epoch is bumped a second time. In
  // particular, if fencing the current producer exhausts the available epochs for the current producerId,
  // then when the client retries, we will generate a new producerId.
  Right(coordinatorEpoch, txnMetadata.prepareFenceProducerEpoch())

最后如果之前的事务处于进行中的状态会回滚事务

handleEndTransaction(transactionalId,
  newMetadata.producerId,
  newMetadata.producerEpoch,
  TransactionResult.ABORT,
  sendRetriableErrorCallback)

或者就是新事务，往事务日志里面插一条日志(对应数据流图中的2a)

txnManager.appendTransactionToLog(transactionalId, coordinatorEpoch, newMetadata, sendPidResponseCallback)

3.客户端调用KafkaProducer.beginTransaction开始新事务。这一步相对简单，就是客户端设置状态成State.IN_TRANSACTION

4.consume-transform-produce过程，这一步是实际消费消息和生产消息的过程。
(1)客户端发送消息时(KafkaProducer.send)，对于新碰到的TopicPartition会触发AddPartitionsToTxnRequest。服务端对应的处理在TransactionCoordinator.handleAddPartitionsToTransaction，主要做的事情是更新事务元数据和记录事务日志(对应数据流图中的4.1a)。在事务中记录partition的作用是后面给事务每个partition发送提交或者回滚标记时需要事务所有的partition。
(2)客户端通过KafkaProducer.send发送消息(ProduceRequest)，比较早的kafka版本增加了PID，epoch，sequence number等几个字段，对应数据流图中的4.2a
(3)客户端调用KafkaProducer.sendOffsetsToTransaction保存事务消费位点。服务端的处理逻辑在TransactionCoordinator.handleAddPartitionsToTransaction，和4.1基本是一样的，不同的是4.3记录的是记录消费位点的topic(GROUP_METADATA_TOPIC_NAME)。
(4)4.3调用的后半部分会触发TxnOffsetCommitRequest，通过数据消息的方式把消费位点持久化到GROUP_METADATA_TOPIC_NAME(__consumer-offsets)这个topic里面去，对应数据流图中的4.4a。
客户端发起逻辑在AddOffsetsToTxnHandler.handleResponse

if (error == Errors.NONE) {
    log.debug("{}Successfully added partition for consumer group {} to transaction", logPrefix,
            builder.consumerGroupId());

    // note the result is not completed until the TxnOffsetCommit returns
    pendingRequests.add(txnOffsetCommitHandler(result, offsets, builder.consumerGroupId()));
    transactionStarted = true;
}

因为需要处理可见性相关的逻辑，服务端事务消费位点和普通消费位点提交的处理逻辑稍有不同，调用GroupCoordinator.handleTxnCommitOffsets而不是handleCommitOffsets。

5.结束事务需要调用KafkaProducer.commitTransaction或者KafkaProducer.abortTransaction
(1)首先客户端会发送一个EndTxnRequest，而服务端由TransactionCoordinator.handleEndTransaction处理。

handleEndTransaction首先会做一个可能的状态转换让事务进入预提交或者预放弃阶段

else txnMetadata.state match {
  case Ongoing =>
    val nextState = if (txnMarkerResult == TransactionResult.COMMIT)
      PrepareCommit
    else
      PrepareAbort
    Right(coordinatorEpoch, txnMetadata.prepareAbortOrCommit(nextState, time.milliseconds()))

接下来会在事务日志里面记录PREPARE_COMMIT或者PREPARE_ABORT日志，对应数据流图中的5.1a

txnManager.appendTransactionToLog(transactionalId, coordinatorEpoch, newMetadata, sendTxnMarkersCallback)

再接下来会往用户数据日志里面发送COMMIT或者ABORT的Control Message，最后往事务日志里面写入COMMIT或者ABORT，才算完成了事务的提交过程。这个过程是用回调的方式组织起来的，代码的流程是TransactionStateManager.appendTransactionToLog->TransactionMarkerChannelManager.addTxnMarkersToSend->TransactionStateManager.appendTransactionToLog

(2)往用户数据日志里面发送COMMIT或者ABORT的Control Message的过程，对应数据流图中的5.2a
发起点在回调方法sendTxnMarkersCallback，这个方法首先会做转台转换让事务进入CompleteCommit或者CompleteAbort状态

case PrepareCommit =>
  if (txnMarkerResult != TransactionResult.COMMIT)
    logInvalidStateTransitionAndReturnError(transactionalId, txnMetadata.state, txnMarkerResult)
  else
    Right(txnMetadata, txnMetadata.prepareComplete(time.milliseconds()))
case PrepareAbort =>
  if (txnMarkerResult != TransactionResult.ABORT)
    logInvalidStateTransitionAndReturnError(transactionalId, txnMetadata.state, txnMarkerResult)
  else
    Right(txnMetadata, txnMetadata.prepareComplete(time.milliseconds()))

最后会往事务相关的每个broker发送WriteTxnMarkersRequest，如果事务包含消费位点也会往__consumer-offsets所在的broker发请求。 broker端的处理在KafkaApis.handleWriteTxnMarkersRequest会把control message写入日志

replicaManager.appendRecords(
  timeout = config.requestTimeoutMs.toLong,
  requiredAcks = -1,
  internalTopicsAllowed = true,
  isFromClient = false,
  entriesPerPartition = controlRecords,
  responseCallback = maybeSendResponseCallback(producerId, marker.transactionResult))

(3)事务日志写入最终的COMMIT或者ABORT日志，对应数据流图的5.3，这一步完成了一个事务就算彻底完成了。
发起点在回调方法appendToLogCallback

精确一次投递

大致列一下流程中的关键节点

1.在客户端每次发送消息之前会检查是否有producerId如果没有会找服务端去申请，Sender.run

if (transactionManager != null) {
    if (!transactionManager.isTransactional()) {
        // this is an idempotent producer, so make sure we have a producer id
        maybeWaitForProducerId();
    }

服务端会在TransactionCoordinator.handleInitProducerId处理，前面事务消息提到过

2.在生成消息内容的时候(RecordAccumulator.drain)会获取当前的的sequenceNumber(TransactionManager.sequenceNumber)放到消息体里面。而sequenceNumber的自增是在发送上一批消息返回是触发的(Sender.handleProduceResponse)。

3.broker实际写入消息之前(Log.append)才会对sequenceNumber进行校验，校验的具体逻辑在ProducerStateManager.validateAppend

其他一些技术点

1.sequenceNumber可以设计成每个producer唯一或者更细粒度的对每个topic-partition唯一，topic-partition唯一的好处是对于每个topic-partition sequenceNumber可以设计成连续的，这样broker端可以做更强的校验，比如检查丢消息，kafka使用的就是细粒度的方法，发现sequenceNumber不连续的时候会抛异常OutOfOrderSequenceException

2.发消息(KafkaProducer.doSend)是个异步的过程，但同时提供Future返回值使得在必要的时候可以把异步变成同步等待。kafka也实现了攒消息批量发送的能力(RecordAccumulator.append)，攒消息的存放方式是一个大hash map，key是topic-partition，消息实际刷出和发送在一个单独的线程中执行，调用Sender.sendProducerData。被刷出的消息的判定在RecordAccumulator.ready，主要依据是消息集是否已满或者是否超时。

3.Consumer消费的时候怎样控制事务可见性呢？一个比较直观的方法就是先把事务消息buffer起来，然后遇到提交或者回滚标志的时候做相应的处理，kafka处理的更巧妙一些。首先是不能读到未提交的事务的控制，kafka引入了lastStableOffset这个概念，lastStableOffset是当前已经提交的事务的最大位点。在ReplicaManager.readFromLocalLog里面有控制，

val initialHighWatermark = localReplica.highWatermark.messageOffset
val lastStableOffset = if (isolationLevel == IsolationLevel.READ_COMMITTED)
  Some(localReplica.lastStableOffset.messageOffset)
else
  None

// decide whether to only fetch committed data (i.e. messages below high watermark)
val maxOffsetOpt = if (readOnlyCommitted)
  Some(lastStableOffset.getOrElse(initialHighWatermark))
else
  None
...
val fetch = log.read(offset, adjustedFetchSize, maxOffsetOpt, minOneMessage, isolationLevel)

这样未提交的事务对客户端就不可见了

4.还有一个需求是要识别并且跳过那些在aborted事务内的消息，这些消息可能和非事务消息混在一起。kafka读消息的返回信息中会带上本批读取的消息中回滚事务列表来帮助客户端跳过。

case class FetchDataInfo(fetchOffsetMetadata: LogOffsetMetadata,
                         records: Records,
                         firstEntryIncomplete: Boolean = false,
                         abortedTransactions: Option[List[AbortedTransaction]] = None)

回滚事务列表是在读取消息日志(Log.read)的过程中撸的

return isolationLevel match {
  case IsolationLevel.READ_UNCOMMITTED => fetchInfo
  case IsolationLevel.READ_COMMITTED => addAbortedTransactions(startOffset, segmentEntry, fetchInfo)
}

那接下来的问题是kafka是如何快速拿到回滚事务列表的呢？kafka为这件事做了一个文件索引，文件后缀名是'.txnindex'，相关的管理逻辑在TransactionIndex。

5.broker端对于producer的状态管理，broker需要记录producer对应的最大sequenceNumber，epoch之类的信息。相关逻辑是放在ProducerStateManager里面的，broker每次写入消息的时候(Log.append)都会更新producer信息(ProducerStateManager.update)。由于只有当有消息写入的时候producer state才会被更新，所以当broker挂掉的时候producer的状态需要被持久化，kafka又弄了一个文件'.snapshot'来持久化producer信息。

设计上的体会

感觉kafka在设计上概念的统一和架构的连贯上做的特别好，比如producerId的引入把精确一次投递和事务消息都给串联起来了。

印象更深刻的例子是kafka早先几个版本依次推出了几个特性，
1.把位点当做普通消息保存
2.加入了消息清理机制，只保留key最新的value
3.加入了broker端的coordinator解决惊群和脑裂问题

然后这几个特性在事务消息这块都用上了，首先把位点当普通消息保存在概念上统一了消息发送和消费，同时消息同步也成了broker之间同步状态的基础机制这样就不用再弄一套状态同步机制了，不过这样做的缺点是只有写消息才能同步broker状态某些特殊情况可能有点小麻烦。利用消息队列保存状态的一个毛病是比较浪费资源，而消息清理机制恰好解决了这个问题。最后是broker端的coordinator机制可以用在consumer group协调者也可以用在事务协调者上面。这种层次递进的特性累加真是相当有美感而且感觉是深思熟虑的结果。

参考资料

https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging

https://cwiki.apache.org/confluence/display/KAFKA/Idempotent+Producer

https://cwiki.apache.org/confluence/display/KAFKA/Transactional+Messaging+in+Kafka
https://medium.com/@jaykreps/exactly-once-support-in-apache-kafka-55e1fdd0a35f

捡贝壳

2017年10月9日星期一

kafka精确一次投递和事务消息学习整理

概述

需求场景

精确一次投递

事务消息

几个关键概念和推导

重要的类图

架构和数据流

事务消息

精确一次投递

其他一些技术点

设计上的体会

参考资料

1 条评论: