[jira] [Created] (KAFKA-5610) KafkaApis.handleWriteTxnMarkerRequest can return UNSUPPORTED_FOR_MESSAGE_FORMAT error on partition emigration

["Apurva Mehta (JIRA)" <ji...@apache.org>]

Apurva Mehta created KAFKA-5610:
-----------------------------------

             Summary: KafkaApis.handleWriteTxnMarkerRequest can return UNSUPPORTED_FOR_MESSAGE_FORMAT error on partition emigration
                 Key: KAFKA-5610
                 URL: https://issues.apache.org/jira/browse/KAFKA-5610
             Project: Kafka
          Issue Type: Bug
    Affects Versions: 0.11.0.0
            Reporter: Apurva Mehta
            Assignee: Apurva Mehta
            Priority: Critical
             Fix For: 0.11.0.1


This bug was revealed by the following system test failure http://confluent-systest.s3-website-us-west-2.amazonaws.com/confluent-kafka-system-test-results/?prefix=2017-07-18--001.1500383975--apache--trunk--28c83d9/

What happened was that a commit marker to the offsets topic (sent as part of the {{producer.sendOffsetsToTransaction}} method) was lost, causing data to be reprocessed, and hence causing the test to fail. 

The bug is that the wrong error code is returned from the handleWriteTxnMarker request when there is partition emigration. In particular, we have: 

{code:scala}
for (marker <- markers.asScala) {
      val producerId = marker.producerId
      val (goodPartitions, partitionsWithIncorrectMessageFormat) = marker.partitions.asScala.partition { partition =>
        replicaManager.getMagic(partition) match {
          case Some(magic) if magic >= RecordBatch.MAGIC_VALUE_V2 => true
          case _ => false
        }
      }

      if (partitionsWithIncorrectMessageFormat.nonEmpty) {
        val currentErrors = new ConcurrentHashMap[TopicPartition, Errors]()
        partitionsWithIncorrectMessageFormat.foreach { partition => currentErrors.put(partition, Errors.UNSUPPORTED_FOR_MESSAGE_FORMAT) }
        updateErrors(producerId, currentErrors)
      }
{code}

But the {{replicaManager.getMagic()}} call will return {{None}} when the partition emigrates, causing the {{handleWriteTxnMarkersRequest}} to return an {{UNSUPPORTED_FOR_MESSAGE_FORMAT}} error. 

From the log, we see that the partition did emigrate a few milliseconds before the {{WriteTxnMarkerRequest}} was sent.

On the old leader, worker10:
{noformat}
./worker10/debug/server.log:32245:[2017-07-18 05:43:20,950] INFO [GroupCoordinator 2]: Unloading group metadata for transactions-test-consumer-group with generation 0 (kafka.coordinator.group.GroupCoordinator)
{noformat}

On the client: 
{noformat}
[2017-07-18 05:43:20,959] INFO [Transaction Marker Request Completion Handler 1]: Sending my-first-transactional-id's transaction marker from partition __consumer_offsets-47 has failed with  UNSUPPORTED_FOR_MESSAGE_FORMAT. This partition will be removed from the set of partitions waiting for completion (kafka.coordinator.transaction.TransactionMarkerRequestCompletionHandler)
{noformat}

As you can see, the client received the response 9 ms after the emigration was initiated on the server.

Since it is perfectly acceptable for the LeaderISR metadata to be propagated asynchronously, we should have more robust handling of emgiration in KafkaApis so that it returns the right error code when handling a request for a partition for which the broker is no longer the leader.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)