今天發(fā)現(xiàn)一臺(tái)kafka broker宕掉，重啟kafka broker集群發(fā)現(xiàn)日志中報(bào)如下錯(cuò)誤，查閱各種資料，解決問(wèn)題如下

一、出現(xiàn)的問(wèn)題

Found a corrupted index file due to requirement failed: Corrupt index found, index file (/var/local/kafka/data/uws_topic-2/00000000000000254473.index) has non-zero size but the last offset is 254473 which is no larger than the base offset 254473.}. deleting /var/local/kafka/data/uws_topic-2/00000000000000254473.timeindex, /var/local/kafka/data/uws_topic-2/00000000000000254473.index and rebuilding index..

發(fā)現(xiàn)損壞的索引文件，在00000000000000254473.index中發(fā)現(xiàn)損壞的索引，索引的大小不是零，但是最后的offset是254473 ， 不大于基偏移量254473，需要?jiǎng)h除254473.timeindex和254473.index，從新構(gòu)建索引。

最后的offset是254473，基偏移量254473，最后的offset與基偏移量相等，二者相減等零才對(duì)，但相減索引的長(zhǎng)度不是零，應(yīng)該是零才對(duì)，所以判斷索引就損壞了

另外附上一張kafka每個(gè)broker工作正常的圖片（圖中為3個(gè)broker、6個(gè)partition、3個(gè)replica）

二、解決問(wèn)題
根據(jù)日志提示，我們需要手動(dòng)刪除每個(gè)partition下的 兩個(gè)索引文件， 重新啟動(dòng)集群，kafka會(huì)自動(dòng)重建索引文件

find /var/local/kafka -name "*.index" |xargs rm -f
find /var/local/kafka -name "*.timeindex" |xargs rm -f

三、出現(xiàn)的原因
出現(xiàn)此問(wèn)題一般情況是Kafka Brokenr非正常關(guān)閉停止，導(dǎo)致索引損壞

因此需要了解一下Kafka log的格式及啟動(dòng)時(shí)候加載過(guò)程

Kafka log Segment的index文件格式：
首先需要指出，.index文件不是為每個(gè)消息都指定到物理位置的映射；舉個(gè)例子，假設(shè)有20000~20009共10條消息，.index文件可配置為每條entry
指定連續(xù)10條消息的物理位置映射，該例中，index entry會(huì)記錄偏移量為20000的消息到其物理文件位置，一旦該條消息被定位，20001~20009可以很快查到。
每個(gè)entry大小8字節(jié)，定義了消息和其在文件中物理位置的映射；
前4個(gè)字節(jié)是這個(gè)消息相對(duì)于該log segment第一個(gè)消息offset的相對(duì)偏移量，后4個(gè)字節(jié)是這個(gè)消息在log文件中的物理位置；
Kafka啟動(dòng)時(shí)加載log的步驟：
以一個(gè)partition log目錄為例:：

<1> 首先刪除所有后綴名為.cleaned和.delete的文件：
<2> 對(duì)于.swp結(jié)尾的文件，如果是log文件則直接恢復(fù)(去掉.swp, 變?yōu)?log)；
如果是index文件直接刪掉（然后rebuild index文件）；
<3> 對(duì)于.index文件，如果沒(méi)有對(duì)應(yīng)的.log文件(同一個(gè)logSement其index和log的主文件名相同), 則刪除該index文件；
<4> 對(duì)于.log文件，加載如內(nèi)存；如果其沒(méi)有對(duì)應(yīng)的.index文件（可能在第<2>步中被刪除), 重新恢復(fù)其index文件；
<5> 假設(shè)到這一步為止Kafka還沒(méi)有加載到logSements, 說(shuō)明該partition log目錄下為空，一個(gè)新的log sement對(duì)象會(huì)被創(chuàng)建在內(nèi)存；
反之則轉(zhuǎn)向第<6>步；
<6> 如果Kafka已經(jīng)加載到log, 最會(huì)開(kāi)始recover log segments。

至于為什么要recover log segments, 是因?yàn)榇蠖鄶?shù)情況下，recover的目的就是檢查Kafka上次關(guān)閉時(shí)是不是cleanShutDown （可通過(guò)檢查partition log目錄下是不是有后綴名為.kafka_cleanshutdown的文件確定）；
如果是cleanShutDown(后綴名為.kafka_cleanshutDown的文件存在），則無(wú)需recover log segment；
如果不是cleanShutDown, 則需要recover log segments；
<6.1> 這里解釋下什么是recover a log segment?
在非cleanShutDown情況下， 一個(gè)log sement的log及index文件末尾可能有一些不合法的數(shù)據(jù)(invalid), 我們需要把它們截掉；
首先要做的最簡(jiǎn)單檢查，是log或index文件大小不能超過(guò)配置中設(shè)定的值（比方說(shuō)一個(gè).log文件中被設(shè)定最多保存10000條消息，超過(guò)10000條的都要拋棄掉）；
<7> 最后做sanityCheck, 主要是檢查每個(gè)log sement的index文件，確保不會(huì)加載一個(gè)出錯(cuò)的Log Segment；