kafka-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Sriram Subramanian (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (KAFKA-772) System Test Transient Failure on testcase_0122
Date Tue, 05 Mar 2013 00:31:13 GMT

    [ https://issues.apache.org/jira/browse/KAFKA-772?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13592850#comment-13592850
] 

Sriram Subramanian commented on KAFKA-772:
------------------------------------------

The test failed on Monday and then again failed on Friday. It was clear that the issue was
timing related. We tried to reproduce the failure on the local box (repeatedly running the
test) but could not reproduce it. I did some code browsing but did not have much luck. So
I decided to setup tracing and run the test repeatedly in a distributed environment over the
weekend and was hoping that it would fail. Luckily, it did and the trace logs proved to be
useful in identifying the issue. Thanks to John for setting this up.

What you see below are excerpts from the trace log which pertain to this failure at different
points in time. In this particular failure, topic_2 / partitions 2 had missing logical offsets
from 570 to 582 on broker 3 (3 brokers in total).

current fetch offset = 582 
current HW = 570
Leader for topic_2/partition 2 = broker 2

1. The lines below show the Fetch request that was issued by broker 3 to broker 2 just before
broker 1 was shutdown. The requested offset is 582 for [test_2,2].

[2013-03-02 12:37:56,034] TRACE [ReplicaFetcherThread-0-2], issuing to broker 2 of fetch request
Name: FetchRequest; Version: 0; CorrelationId: 121; ClientId: ReplicaFetcherThread-0-2; ReplicaId:
3; MaxWait: 500 ms; MinBytes: 4096 bytes; RequestInfo: [test_1,0] -> PartitionFetchInfo(700,1048576),[test_2,1]
-> PartitionFetchInfo(677,1048576),[test_2,2] -> PartitionFetchInfo(582,1048576),[test_2,0]
-> PartitionFetchInfo(679,1048576),[test_1,2] -> PartitionFetchInfo(600,1048576),[test_1,1]
-> PartitionFetchInfo(699,1048576) (kafka.server.ReplicaFetcherThread)

2. Broker 1 is shutdown and broker 3 handles leader and isr request. Note that [test_2,2]
still follows broker 2 but we still issue a makefollower call for it.

[2013-03-02 12:37:56,086] INFO Replica Manager on Broker 3: Handling leader and isr request
Name: LeaderAndIsrRequest; Version: 0; CorrelationId: 2; ClientId: ; AckTimeoutMs: 1000 ms;
ControllerEpoch: 2; PartitionStateInfo: (test_1,0) -> PartitionStateInfo(LeaderIsrAndControllerEpoch({
"ISR":"2,1,3", "leader":"2", "leaderEpoch":"1" },1),3),(test_2,1) -> PartitionStateInfo(LeaderIsrAndControllerEpoch({
"ISR":"2,3", "leader":"2", "leaderEpoch":"2" },2),3),(test_2,2) -> PartitionStateInfo(LeaderIsrAndControllerEpoch({
"ISR":"2,1,3", "leader":"2", "leaderEpoch":"1" },1),3),(test_2,0) -> PartitionStateInfo(LeaderIsrAndControllerEpoch({
"ISR":"2,3", "leader":"2", "leaderEpoch":"2" },2),3),(test_1,2) -> PartitionStateInfo(LeaderIsrAndControllerEpoch({
"ISR":"2,3", "leader":"2", "leaderEpoch":"2" },2),3),(test_1,1) -> PartitionStateInfo(LeaderIsrAndControllerEpoch({
"ISR":"2,1,3", "leader":"2", "leaderEpoch":"1" },1),3); Leaders: id:2,host:xxxx(kafka.server.ReplicaManager)

3. The leader and isr request results in removing the fetcher to broker 2 for [test_2,2],
truncating the log to high watermark (570) and then adding back the fetcher to the same broker.

[2013-03-02 12:37:56,088] INFO [ReplicaFetcherManager on broker 3] removing fetcher on topic
test_2, partition 2 (kafka.server.ReplicaFetcherManager)
[2013-03-02 12:37:56,088] INFO [Kafka Log on Broker 3], Truncated log segment /tmp/kafka_server_3_logs/test_2-2/00000000000000000000.log
to target offset 570 (kafka.log.Log)
[2013-03-02 12:37:56,088] INFO [ReplicaFetcherManager on broker 3] adding fetcher on topic
test_2, partion 2, initOffset 570 to broker 2 with fetcherId 0 (kafka.server.ReplicaFetcherManager)

4. The leader and isr request is completed at this point of time.

[2013-03-02 12:37:56,090] INFO Replica Manager on Broker 3: Completed leader and isr request
Name: LeaderAndIsrRequest; Version: 0; CorrelationId: 2; ClientId: ; AckTimeoutMs: 1000 ms;
ControllerEpoch: 2; PartitionStateInfo: (test_1,0) -> PartitionStateInfo(LeaderIsrAndControllerEpoch({
"ISR":"2,1,3", "leader":"2", "leaderEpoch":"1" },1),3),(test_2,1) -> PartitionStateInfo(LeaderIsrAndControllerEpoch({
"ISR":"2,3", "leader":"2", "leaderEpoch":"2" },2),3),(test_2,2) -> PartitionStateInfo(LeaderIsrAndControllerEpoch({
"ISR":"2,1,3", "leader":"2", "leaderEpoch":"1" },1),3),(test_2,0) -> PartitionStateInfo(LeaderIsrAndControllerEpoch({
"ISR":"2,3", "leader":"2", "leaderEpoch":"2" },2),3),(test_1,2) -> PartitionStateInfo(LeaderIsrAndControllerEpoch({
"ISR":"2,3", "leader":"2", "leaderEpoch":"2" },2),3),(test_1,1) -> PartitionStateInfo(LeaderIsrAndControllerEpoch({
"ISR":"2,1,3", "leader":"2", "leaderEpoch":"1" },1),3); Leaders: id:2,host:xxxx (kafka.server.ReplicaManager)


5.  A log append happens at offset 582 though the nextOffset for the log is at 570. This append
actually pertains to the fetch request at step 1. This explains the gap in the log.

[2013-03-02 12:37:56,098] TRACE [Kafka Log on Broker 3], Appending message set to test_2-2
offset: 582 nextOffset: 570 messageSet: ByteBufferMessageSet(MessageAndOffset(Message(magic
= 0, attributes = 0, crc = 1408289663, key = null, payload = java.nio.HeapByteBuffer[pos=0
lim=500 cap=500]),582), MessageAndOffset(Message(magic = 0, attributes = 0, crc = 3696400058,
key = null, payload = java.nio.HeapByteBuffer[pos=0 lim=500 cap=500]),583), MessageAndOffset(Message(magic
= 0, attributes = 0, crc = 2403920749, key = null, payload = java.nio.HeapByteBuffer[pos=0
lim=500 cap=500]),584), ) (kafka.log.Log)

>From the set of steps above, it is clear that some thing is causing the fetch request
at step 1 to complete even though step 2 and 3 removed the fetcher for that topic,partition.

Looking at the code now it becomes obvious. The race condition is between the thread that
removes the fetcher, truncates the log and adds the fetcher back and the thread that fetches
bytes from the leader. Follow the steps below to understand what is happening.

Partition.Scala

          replicaFetcherManager.removeFetcher(topic, partitionId)           --> step 2
: Removes the topic,partition – offset mapping from partitionMap in AbstractFetcherThread
          // make sure local replica exists
          val localReplica = getOrCreateReplica()
          localReplica.log.get.truncateTo(localReplica.highWatermark)    --> step 3 : Truncates
to offset 570
          inSyncReplicas = Set.empty[Replica]
          leaderEpoch = leaderAndIsr.leaderEpoch
          zkVersion = leaderAndIsr.zkVersion
          leaderReplicaIdOpt = Some(newLeaderBrokerId)
          // start fetcher thread to current leader
          replicaFetcherManager.addFetcher(topic, partitionId, localReplica.logEndOffset,
leaderBroker)    --> step 4: Sets the new fetcher to fetch from the log end offset which
is at 570 at this point

AbstractFetcherThread.Scala

private def processFetchRequest(fetchRequest: FetchRequest) {
    val partitionsWithError = new mutable.HashSet[TopicAndPartition]
    var response: FetchResponse = null
    try {
      trace("issuing to broker %d of fetch request %s".format(sourceBroker.id, fetchRequest))
      response = simpleConsumer.fetch(fetchRequest)
    } catch {
      case t =>
        debug("error in fetch %s".format(fetchRequest), t)
        if (isRunning.get) {
          partitionMapLock synchronized {
            partitionsWithError ++= partitionMap.keys
          }
        }
    }
    fetcherStats.requestRate.mark()   -->  step 1 : Fetch completes. Fetch request is from
offset 582.

    if (response != null) {
      // process fetched data 
      partitionMapLock.lock()     ---> step 5: This is where the fetch request is waiting
when the addFetcher in Partition.Scala is executing above
      try {
        response.data.foreach {
          case(topicAndPartition, partitionData) =>
            val (topic, partitionId) = topicAndPartition.asTuple
            val currentOffset = partitionMap.get(topicAndPartition)
            if (currentOffset.isDefined) {
              partitionData.error match {
                case ErrorMapping.NoError =>
                  val messages = partitionData.messages.asInstanceOf[ByteBufferMessageSet]
                  val validBytes = messages.validBytes
                  val newOffset = messages.lastOption match {          -->  step 6: The
newOffset is set to 587 and partitionMap is updated
                    case Some(m: MessageAndOffset) => m.nextOffset
                    case None => currentOffset.get
                  }
                  partitionMap.put(topicAndPartition, newOffset)
                  fetcherLagStats.getFetcherLagStats(topic, partitionId).lag = partitionData.hw
- newOffset
                  fetcherStats.byteRate.mark(validBytes)
                  // Once we hand off the partition data to the subclass, we can't mess with
it any more in this thread
                  processPartitionData(topicAndPartition, currentOffset.get, partitionData)
   --> step 7: This appends data to the log with logical offsets from 582 – 587. Note
that the offset passed to this method is 570 (currentOffset). Hence all offset validation
checks in processPartitionData passes.
                case ErrorMapping.OffsetOutOfRangeCode =>
                  try {
                    val newOffset = handleOffsetOutOfRange(topicAndPartition)
                    partitionMap.put(topicAndPartition, newOffset)
                    warn("current offset %d for topic %s partition %d out of range; reset
offset to %d"
                      .format(currentOffset.get, topic, partitionId, newOffset))
                  } catch {
                    case e =>
                      warn("error getting offset for %s %d to broker %d".format(topic, partitionId,
sourceBroker.id), e)
                      partitionsWithError += topicAndPartition
                  }
                case _ =>
                  warn("error for %s %d to broker %d".format(topic, partitionId, sourceBroker.id),
                    ErrorMapping.exceptionFor(partitionData.error))
                  partitionsWithError += topicAndPartition
              }
            }
        }
      } finally {
        partitionMapLock.unlock()
      }
    }
                
> System Test Transient Failure on testcase_0122
> ----------------------------------------------
>
>                 Key: KAFKA-772
>                 URL: https://issues.apache.org/jira/browse/KAFKA-772
>             Project: Kafka
>          Issue Type: Bug
>    Affects Versions: 0.8
>            Reporter: John Fung
>            Assignee: Sriram Subramanian
>              Labels: kafka-0.8, p1
>         Attachments: KAFKA-772.patch, testcase_0122.tar.gz, testcase_0125.tar.gz
>
>
> * This test case is failing randomly in the past few weeks. Please note there is a small
% data loss allowance for the test case with Ack = 1. But the failure in this case is the
mismatch of log segment checksum across the replicas.
> * Test description:
> 3 brokers cluster
> Replication factor = 3
> No. topic = 2
> No. partitions = 3
> Controlled failure (kill -15)
> Ack = 1
> * Test case output
> _test_case_name  :  testcase_0122
> _test_class_name  :  ReplicaBasicTest
> arg : auto_create_topic  :  true
> arg : bounce_broker  :  true
> arg : broker_type  :  leader
> arg : message_producing_free_time_sec  :  15
> arg : num_iteration  :  3
> arg : num_partition  :  3
> arg : replica_factor  :  3
> arg : sleep_seconds_between_producer_calls  :  1
> validation_status  : 
>      Leader Election Latency - iter 1 brokerid 3  :  377.00 ms
>      Leader Election Latency - iter 2 brokerid 1  :  374.00 ms
>      Leader Election Latency - iter 3 brokerid 2  :  384.00 ms
>      Leader Election Latency MAX  :  384.00
>      Leader Election Latency MIN  :  374.00
>      Unique messages from consumer on [test_1] at simple_consumer_test_1-0_r1.log  :
 1750
>      Unique messages from consumer on [test_1] at simple_consumer_test_1-0_r2.log  :
 1750
>      Unique messages from consumer on [test_1] at simple_consumer_test_1-0_r3.log  :
 1750
>      Unique messages from consumer on [test_1] at simple_consumer_test_1-1_r1.log  :
 1750
>      Unique messages from consumer on [test_1] at simple_consumer_test_1-1_r2.log  :
 1750
>      Unique messages from consumer on [test_1] at simple_consumer_test_1-1_r3.log  :
 1750
>      Unique messages from consumer on [test_1] at simple_consumer_test_1-2_r1.log  :
 1500
>      Unique messages from consumer on [test_1] at simple_consumer_test_1-2_r2.log  :
 1500
>      Unique messages from consumer on [test_1] at simple_consumer_test_1-2_r3.log  :
 1500
>      Unique messages from consumer on [test_2]  :  5000
>      Unique messages from consumer on [test_2] at simple_consumer_test_2-0_r1.log  :
 1714
>      Unique messages from consumer on [test_2] at simple_consumer_test_2-0_r2.log  :
 1714
>      Unique messages from consumer on [test_2] at simple_consumer_test_2-0_r3.log  :
 1680
>      Unique messages from consumer on [test_2] at simple_consumer_test_2-1_r1.log  :
 1708
>      Unique messages from consumer on [test_2] at simple_consumer_test_2-1_r2.log  :
 1708
>      Unique messages from consumer on [test_2] at simple_consumer_test_2-1_r3.log  :
 1708
>      Unique messages from consumer on [test_2] at simple_consumer_test_2-2_r1.log  :
 1469
>      Unique messages from consumer on [test_2] at simple_consumer_test_2-2_r2.log  :
 1469
>      Unique messages from consumer on [test_2] at simple_consumer_test_2-2_r3.log  :
 1469
>      Unique messages from producer on [test_2]  :  4900
>      Validate for data matched on topic [test_1] across replicas  :  PASSED
>      Validate for data matched on topic [test_2]  :  FAILED
>      Validate for data matched on topic [test_2] across replicas  :  FAILED
>      Validate for merged log segment checksum in cluster [source]  :  FAILED
>      Validate leader election successful  :  PASSED

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira

Mime
View raw message