背景
首先，交代一下背景胚股。
團(tuán)隊最近在做kafka預(yù)發(fā)環(huán)境與生產(chǎn)環(huán)境隔離，同時由于原來是基于原生的kafka client自行封裝來支持多套kafka集群且多線程并行消費的邏輯裙秋，以及切換集群IP到對應(yīng)的L5上琅拌，所以除了切換集群外，還需要更換SDK
1摘刑、kafka 環(huán)境隔離是非常簡單的事进宝，切換集群IP即可。
2枷恕、更換SDK涉及很多代碼改造党晋，且SDK還需要支持L5，兼容IP的問題徐块。由于這是一個共性問題隶校，所以SDK由支撐團(tuán)隊來開發(fā)。
以下簡單過一下這個過程遇到的幾個問題蛹锰。由于個人覺得比較有代表性深胳，故而記錄一下。
問題1铜犬、SpringBoot版本不一
這是一個極其容易被忽視舞终，但是其實非常容易出現(xiàn)的問題轻庆，目前部門有5-6個小組在做研發(fā)，如果沒有在最開始就做版本統(tǒng)一管理敛劝，及其容易出現(xiàn)版本不一致的問題余爆。引包之后，本地運行出現(xiàn)如下情況：
看圖我并沒有第一時間懷疑版本問題夸盟，還以為是不是本地Maven又有點毛病蛾方，倒騰了幾次實在沒招，點開同事提供的SDK包代碼翻起來上陕，翻看時才發(fā)現(xiàn)問題桩砰。
低版本的（2.0.3）該接口沒有default實現(xiàn)，而在高版本（2.2.5）上是有默認(rèn)default的實現(xiàn)释簿，而SDK由于是基于高版本開發(fā)所以只實現(xiàn)了必要的方法亚隅。此時回頭去看的異常日志時，確實可以對應(yīng)起來庶溶。
問題二煮纵、發(fā)現(xiàn)相同的消費者與topic在不同的環(huán)境下居然使用不同的group.id
在發(fā)現(xiàn)這個問題時，難以置信偏螺。
由于同事提供的SDK相關(guān)使用文檔上是使用@KafkaListener(topics='xxx', groupId='yyy', containerFactory='zzz')的方式行疏，聯(lián)想到注解是一種編譯期元數(shù)據(jù)，第一個錯誤的想法是排除掉可配置問題套像，于是想到的解決方案是采用新的groupId來統(tǒng)一不同環(huán)境的差異酿联。（習(xí)慣性思維帶來的可怕后果是繞過了正確的方式）
更換groupId會帶來什么問題？
允許更換groupId的前置條件必須是不能丟失未消費的消息凉夯。那么必須了解一下kafka的消費進(jìn)度管理的基本原理货葬，消費進(jìn)度與groupId具備什么相關(guān)性嗎采幌？這里會涉及到幾個概念性的問題劲够，比如consumer group、offset休傍、__consumer_offset topic等等征绎，這些概念很輕易可以在網(wǎng)絡(luò)上搜到，自行補充磨取。
如下圖人柿，很好的說明了groupId 與 offset的關(guān)聯(lián)關(guān)系。
那么忙厌，如果更換groupId 會出現(xiàn)什么問題凫岖？意味著該consumer client并沒有相關(guān)的消費進(jìn)度，客戶端更換groupId重新啟動后逢净，從服務(wù)端獲取不到對應(yīng)的offset時哥放，有一個參數(shù)會指導(dǎo)服務(wù)端如何分配offset到客戶端auto.offset.reset歼指，值包括earliest、lastest甥雕。丟失了offset踩身，如果采用earliest，那就相當(dāng)于把消息都撈出來重新消費社露，這肯定不行挟阻，那么lastest會拿到一個什么參數(shù)呢？從網(wǎng)絡(luò)上去找峭弟，很多說會消費最新的消息附鸽，那么到底什么是最新的消息，又沒有提及孟害，后來看到一篇文檔說會從hw處開始消費拒炎，翻了一會源碼確實找到一個看起來有點意思的地方。
def legacyFetchOffsetsForTimestamp(timestamp: Long,
maxNumOffsets: Int,
isFromConsumer: Boolean,
fetchOnlyFromLeader: Boolean): Seq[Long] = inReadLock(leaderIsrUpdateLock) {
val localLog = localLogWithEpochOrException(Optional.empty(), fetchOnlyFromLeader)
val allOffsets = localLog.legacyFetchOffsetsBefore(timestamp, maxNumOffsets)
if (!isFromConsumer) {
allOffsets
} else {
val hw = localLog.highWatermark // 高水位
if (allOffsets.exists(_ > hw)) // 由于對scale相關(guān)語法并不了解挨务，所以不確定最終返回的offset具體是多少击你，只能說猜測也許是高水位
hw +: allOffsets.dropWhile(_ > hw)
else
allOffsets
}
}
那么這個highWatermark又是如何更新的？
private def maybeIncrementLeaderHW(leaderLog: Log, curTime: Long = time.milliseconds): Boolean = {
inReadLock(leaderIsrUpdateLock) {
// maybeIncrementLeaderHW is in the hot path, the following code is written to
// avoid unnecessary collection generation
var newHighWatermark = leaderLog.logEndOffsetMetadata
remoteReplicasMap.values.foreach { replica =>
// flower副本的LEO 小于 leader副本LEO && （上一次拉取的時間距現(xiàn)在還在最大拉取時間之內(nèi) 或者 這是一個ISR副本）
if (replica.logEndOffsetMetadata.messageOffset < newHighWatermark.messageOffset &&
(curTime - replica.lastCaughtUpTimeMs <= replicaLagTimeMaxMs || inSyncReplicaIds.contains(replica.brokerId))) {
newHighWatermark = replica.logEndOffsetMetadata // 新水位為flower副本LEO
}
} // 由于newHighWatermark在循環(huán)中更新谎柄，那么最終newHighWatermark = min(replica.logEndOffsetMetadata.messageOffset)
// ....
}
}
可以理解為hw指的是：多個isr flower副本已經(jīng)更新到的最新消息位置中最小的一個丁侄，即min(replica.logEndOffsetMetadata.messageOffset)
高位水hw是客戶端消費者可見的最新的消息，那么假定原來的groupId 消費者的進(jìn)度記錄為offset1朝巫，如果更換了groupId后消費的 offset2 = hw鸿摇，那么潛在丟失的消息包括offset_missing = hw - offset1.。所以更換groupId是風(fēng)險很高的操作劈猿。
GroupId 可配置問題
峰回路轉(zhuǎn)拙吉，發(fā)現(xiàn)雖然 @KafakListener是注解，但是依舊可以配合配置項來完成消費者的初始化揪荣。
1筷黔、可以通過@KafakListener(groupId = '${xxx}')來完成注入（是嘗試行為，結(jié)果發(fā)現(xiàn)可行）
2仗颈、通過 spring.kafka.consumer.group-id來實現(xiàn)配置注入（看配置猜測也許可以）
注意佛舱，使用第二種方式時，需要在@KafakListener去掉 id挨决、groupId兩個字段的配置请祖，為什么id會影響groupId，多少有點出乎意料脖祈。
首先看代碼KafkaListenerAnnotationBeanPostProcessor
private String getEndpointGroupId(KafkaListener kafkaListener, String id) {
String groupId = null;
if (StringUtils.hasText(kafkaListener.groupId())) {
groupId = resolveExpressionAsString(kafkaListener.groupId(), "groupId");
}
// 沒有配置groupId肆捕，但是配置了id時，groupId = id, idIsGroup默認(rèn)是true,原來沒注意居然還有這么個參數(shù)
if (groupId == null && kafkaListener.idIsGroup() && StringUtils.hasText(kafkaListener.id())) {
groupId = id;
}
return groupId;
}
在KafkaMessageListenerContainer中有如下判斷
private final String consumerGroupId = this.containerProperties.getGroupId() == null // this.containerProperties.getGroupId() 就是從注解中獲取的數(shù)據(jù)
? (String) KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties()
.get(ConsumerConfig.GROUP_ID_CONFIG) // 從配置文件中讀取到的配置
: this.containerProperties.getGroupId();
總結(jié)
為什么前文說這是比較有代表性的問題呢盖高？
首先我覺得版本不一致這個問題慎陵，有時候因為種種原因（特別是大家一個團(tuán)隊內(nèi)）你真的很難說第一眼就發(fā)現(xiàn)這種不一致掏秩，然后你花費了很多時間去排查最終得出一個沒多少價值的結(jié)果，因為它也真的是很容易避免荆姆。其次后邊遇到這種groupId問題蒙幻。如果想當(dāng)然的認(rèn)為配置會一樣，于是最終上線了發(fā)現(xiàn)消息丟失了那真的是粗心大意就把你埋了胆筒。如果從一開始就去翻對應(yīng)的文檔邮破，也許就很好的明白可配置，以及idIsGroup的細(xì)節(jié)了仆救，壓根不需要去看源代碼抒和。當(dāng)然，groupId問題進(jìn)而引導(dǎo)開發(fā)同學(xué)去排查整個思路彤蔽，對于開發(fā)同學(xué)掌握中間件有挺大幫助的摧莽，無目的性的看代碼容易遺忘，但是與實踐相結(jié)合才能讓你明白的更加深刻顿痪。
