kafka-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Blake Smith <blake.sm...@tempo-db.com>
Subject Topic corruption from hardware failure (0.7.1)
Date Tue, 16 Jul 2013 20:10:09 GMT
Hi everyone,

Last week, one of our production Kafka 0.7.1 servers had a hardware failure
that resulted in an unclean restart. When the server came back up 5 minutes
later, there were two topic corruption problems that we had to handle to
get the pipeline working again.

1. The kafka log directory had malformed partition directory names [1]. The
partition directories were shortened names of our topic naming scheme.
Inside some of these directories was a 0 byte .kafka file. This prevented
the kafka server process from starting. To resolve this, we manually
removed the malformed topic directories and the empty log files.

Is this a known issue? If so, has it been addressed in 0.8? This seems like
a bug.

2. One of the partitions had a corrupt log message (crc32 check failure)
that prevented consumers from advancing on that partition [2]. To resolve
this issue we had to hexdump the log file seeking to the byte offset
provided by the consumer exception and remove the corrupt message at the
frame boundary. After removing the corrupt message, we created a task to
parse the log file and put it back into our producer / consumer pipeline.
The InvalidMessageException never bubbled to our consumer thread,
preventing us from handling this error without manual manipulation of the
log files (our consumer code was catching InvalidMessageException, but it
never reached that point).

Is a corrupt log file a state that Kafka is supposed to handle, or is our
consumer code supposed to handle it? I'm confused about how we can avoid
manually cleaning up log files with a hex editor if we get into this state
again.

Thanks!

Blake

[1] Corrupt partition directories in the broker data directories:

0    [main] INFO  kafka.utils.Utils$  - The number of partitions for
topic  datapoints-write : 50
1    [main] INFO  kafka.utils.Utils$  - The number of partitions for
topic  datapoints-increment : 50
2    [main] INFO  kafka.utils.Utils$  - The number of partitions for
topic  series-index : 5
5    [main] INFO  kafka.server.KafkaServer  - Starting Kafka server...
17   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-write-18'
32   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-write-18/00000000095029068903.kafka in mutable
mode, recovery false
46   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-increment-15'
46   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-increment-15/00000000000000000000.kafka in
mutable mode, recovery false
47   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-increment-25'
47   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-increment-25/00000000000000000000.kafka in
mutable mode, recovery false
47   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-write-27'
48   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-write-27/00000000093417873484.kafka in mutable
mode, recovery false
48   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-write-9'
48   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-write-9/00000000092880448733.kafka in mutable
mode, recovery false
49   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-increment-8'
49   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-increment-8/00000000000000000000.kafka in
mutable mode, recovery false
49   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-increment-29'
49   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-increment-29/00000000008053064625.kafka in
mutable mode, recovery false
50   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-increment-34'
50   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-increment-34/00000000000000000000.kafka in
mutable mode, recovery false
50   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-increment-19'
51   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-increment-19/00000000000000000000.kafka in
mutable mode, recovery false
51   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-increment-16'
51   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-increment-16/00000000000000000000.kafka in
mutable mode, recovery false
51   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-increment-27'
52   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-increment-27/00000000000000000000.kafka in
mutable mode, recovery false
52   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-write-34'
52   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-write-34/00000000092880904403.kafka in mutable
mode, recovery false
53   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-write-21'
53   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-write-21/00000000095028969469.kafka in mutable
mode, recovery false
53   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-write-48'
54   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-write-48/00000000094491048876.kafka in mutable
mode, recovery false
54   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-increment-40'
55   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-increment-40/00000000000000000000.kafka in
mutable mode, recovery false
55   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-increment-6'
55   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-increment-6/00000000000000000000.kafka in
mutable mode, recovery false
56   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-write-25'
56   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-write-25/00000000093417394574.kafka in mutable
mode, recovery false
56   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-write-33'
57   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-write-33/00000000093417473922.kafka in mutable
mode, recovery false
57   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-write-35'
58   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-write-35/00000000092880891040.kafka in mutable
mode, recovery false
58   [main] INFO  kafka.log.LogManager  - Loading log 'datapoints-wr'
58   [main] INFO  kafka.log.Log  - Loading the last segment
/var/kafka/datapoints-wr/00000000000000000000.kafka in mutable mode,
recovery false
60   [main] FATAL kafka.server.KafkaServerStartable  - Fatal error
during KafkaServerStable startup. Prepare to shutdown
java.lang.NumberFormatException: For input string: "wr"
        at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65)
        at java.lang.Integer.parseInt(Integer.java:481)
        at java.lang.Integer.parseInt(Integer.java:514)
        at scala.collection.immutable.StringLike$class.toInt(StringLike.scala:207)
        at scala.collection.immutable.StringOps.toInt(StringOps.scala:31)
        at kafka.utils.Utils$.getTopicPartition(Utils.scala:558)
        at kafka.log.LogManager$$anonfun$4.apply(LogManager.scala:71)
        at kafka.log.LogManager$$anonfun$4.apply(LogManager.scala:65)
        at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:34)
        at scala.collection.mutable.ArrayOps.foreach(ArrayOps.scala:34)
        at kafka.log.LogManager.<init>(LogManager.scala:65)
        at kafka.server.KafkaServer.startup(KafkaServer.scala:58)
        at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:34)
        at kafka.Kafka$.main(Kafka.scala:50)
        at kafka.Kafka.main(Kafka.scala)
61   [main] INFO  kafka.server.KafkaServer  - Shutting down Kafka server
62   [main] INFO  kafka.utils.KafkaScheduler  - shutdown scheduler
kafka-logcleaner-
62   [main] INFO  kafka.server.KafkaServer  - Kafka server shut down completed

[blake@kafka03 kafka]$ ls
dat                      datapoints-increment-40  datapoints-write-26
datapoints-increment-0   datapoints-increment-41  datapoints-write-27
datapoints-increment-1   datapoints-increment-42  datapoints-write-28
datapoints-increment-10  datapoints-increment-43  datapoints-write-29
datapoints-increment-11  datapoints-increment-44  datapoints-write-3
datapoints-increment-12  datapoints-increment-45  datapoints-write-30
datapoints-increment-13  datapoints-increment-46  datapoints-write-31
datapoints-increment-14  datapoints-increment-47  datapoints-write-32
datapoints-increment-15  datapoints-increment-48  datapoints-write-33
datapoints-increment-16  datapoints-increment-49  datapoints-write-34
datapoints-increment-17  datapoints-increment-5   datapoints-write-35
datapoints-increment-18  datapoints-increment-6   datapoints-write-36
datapoints-increment-19  datapoints-increment-7   datapoints-write-37
datapoints-increment-2   datapoints-increment-8   datapoints-write-38
datapoints-increment-20  datapoints-increment-9   datapoints-write-39
datapoints-increment-21  datapoints-wr            datapoints-write-4
datapoints-increment-22  datapoints-writ          datapoints-write-40
datapoints-increment-23  datapoints-write-0       datapoints-write-41
datapoints-increment-24  datapoints-write-1       datapoints-write-42
datapoints-increment-25  datapoints-write-10      datapoints-write-43
datapoints-increment-26  datapoints-write-11      datapoints-write-44
datapoints-increment-27  datapoints-write-12      datapoints-write-45
datapoints-increment-28  datapoints-write-13      datapoints-write-46
datapoints-increment-29  datapoints-write-14      datapoints-write-47
datapoints-increment-3   datapoints-write-15      datapoints-write-48
datapoints-increment-30  datapoints-write-16      datapoints-write-49
datapoints-increment-31  datapoints-write-17      datapoints-write-5
datapoints-increment-32  datapoints-write-18      datapoints-write-6
datapoints-increment-33  datapoints-write-19      datapoints-write-7
datapoints-increment-34  datapoints-write-2       datapoints-write-8
datapoints-increment-35  datapoints-write-20      datapoints-write-9
datapoints-increment-36  datapoints-write-21      series-index-0
datapoints-increment-37  datapoints-write-22      series-index-1
datapoints-increment-38  datapoints-write-23      series-index-2
datapoints-increment-39  datapoints-write-24      series-index-3
datapoints-increment-4   datapoints-write-25      series-index-4
[blake@kafka03 kafka]$ cd datapoints-wr
[blake@kafka03 datapoints-wr]$ ls -lah
total 8.0K
drwxr-xr-x   2 kafka kafka 4.0K Jul 11 18:51 .
drwxr-xr-x 110 kafka kafka 4.0K Jul 11 19:05 ..
-rw-r--r--   1 kafka kafka    0 Jul 11 18:51 00000000000000000000.kafka



[2] In our consumer logs:

013-07-11 21:33:01,136 61867 [FetchRunnable-0] ERROR
kafka.consumer.FetcherRunnable  - error in FetcherRunnable
kafka.message.InvalidMessageException: message is invalid, compression
codec: NoCompressionCodec size: 79 curr offset: 93706191248 init
offset: 93706191165
        at kafka.message.ByteBufferMessageSet$$anon$1.makeNextOuter(ByteBufferMessageSet.scala:130)
        at kafka.message.ByteBufferMessageSet$$anon$1.makeNext(ByteBufferMessageSet.scala:160)
        at kafka.message.ByteBufferMessageSet$$anon$1.makeNext(ByteBufferMessageSet.scala:100)
        at kafka.utils.IteratorTemplate.maybeComputeNext(IteratorTemplate.scala:59)
        at kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:51)
        at kafka.message.ByteBufferMessageSet.shallowValidBytes(ByteBufferMessageSet.scala:64)
        at kafka.message.ByteBufferMessageSet.validBytes(ByteBufferMessageSet.scala:59)
        at kafka.consumer.PartitionTopicInfo.enqueue(PartitionTopicInfo.scala:57)
        at kafka.consumer.FetcherRunnable$$anonfun$run$6.apply(FetcherRunnable.scala:79)
        at kafka.consumer.FetcherRunnable$$anonfun$run$6.apply(FetcherRunnable.scala:65)
        at scala.collection.LinearSeqOptimized$class.foreach(LinearSeqOptimized.scala:59)
        at scala.collection.immutable.List.foreach(List.scala:76)
        at kafka.consumer.FetcherRunnable.run(FetcherRunnable.scala:65)

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message