kafka-報錯問題整理

1、broker掛了  關鍵字LogDirFailureChannel    NoSuchFileException     Shutdown broker because all log dirs in /tmp/kafka-logs have failed

裝的是單機單節點的kafka，運行了一段時間后掛了，回頭看日志如下，查閱了一些問題單，發現這個問題還是很普遍的

[2022-03-28 10:36:38,194] ERROR Failed to clean up log for __consumer_offsets-2 in dir /tmp/kafka-logs due to IOException (kafka.server.LogDirFailureChannel)
java.nio.file.NoSuchFileException: /tmp/kafka-logs/__consumer_offsets-2/00000000000000000000.log
        at sun.nio.fs.UnixException.translateToIOException(UnixException.java:86)
        at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:102)
        at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:107)
        at sun.nio.fs.UnixCopyFile.move(UnixCopyFile.java:409)
        at sun.nio.fs.UnixFileSystemProvider.move(UnixFileSystemProvider.java:262)
        at java.nio.file.Files.move(Files.java:1395)
        at org.apache.kafka.common.utils.Utils.atomicMoveWithFallback(Utils.java:806)
        at org.apache.kafka.common.record.FileRecords.renameTo(FileRecords.java:224)
        at kafka.log.LogSegment.changeFileSuffixes(LogSegment.scala:489)
        at kafka.log.Log.kafka$log$Log$$asyncDeleteSegment(Log.scala:1960)
        at kafka.log.Log$$anonfun$replaceSegments$3.apply(Log.scala:2023)
        at kafka.log.Log$$anonfun$replaceSegments$3.apply(Log.scala:2018)
        at scala.collection.immutable.List.foreach(List.scala:392)
        at kafka.log.Log.replaceSegments(Log.scala:2018)
        at kafka.log.Cleaner.cleanSegments(LogCleaner.scala:582)
        at kafka.log.Cleaner$$anonfun$doClean$4.apply(LogCleaner.scala:512)
        at kafka.log.Cleaner$$anonfun$doClean$4.apply(LogCleaner.scala:511)
        at scala.collection.immutable.List.foreach(List.scala:392)
        at kafka.log.Cleaner.doClean(LogCleaner.scala:511)
        at kafka.log.Cleaner.clean(LogCleaner.scala:489)
        at kafka.log.LogCleaner$CleanerThread.cleanLog(LogCleaner.scala:350)
        at kafka.log.LogCleaner$CleanerThread.cleanFilthiestLog(LogCleaner.scala:319)
        at kafka.log.LogCleaner$CleanerThread.doWork(LogCleaner.scala:300)
        at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:82)
        Suppressed: java.nio.file.NoSuchFileException: /tmp/kafka-logs/__consumer_offsets-2/00000000000000000000.log -> /tmp/kafka-logs/__consumer_offsets-2/00000000000000000000.log.deleted
                at sun.nio.fs.UnixException.translateToIOException(UnixException.java:86)
                at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:102)
                at sun.nio.fs.UnixCopyFile.move(UnixCopyFile.java:396)
                at sun.nio.fs.UnixFileSystemProvider.move(UnixFileSystemProvider.java:262)
                at java.nio.file.Files.move(Files.java:1395)
                at org.apache.kafka.common.utils.Utils.atomicMoveWithFallback(Utils.java:803)
                ... 17 more
..................

[2022-03-28 10:36:39,277] INFO [ReplicaManager broker=0] Broker 0 stopped fetcher for partitions __consumer_offsets-22,logaudit_20220314-0,__consumer_offsets-30,logaudit_20220307-4,logaudit_20220314-2,logaudit_20220314-11,__consumer_offsets-8,logaudit_20220314-7,__consumer_offsets-21,__consumer_offsets-4,__consumer_offsets-27,__consumer_offsets-7,logaudit_20220307-11,__consumer_offsets-9,__consumer_offsets-46,logaudit_20220307-8,__consumer_offsets-25,__consumer_offsets-35,__consumer_offsets-41,__consumer_offsets-33,__consumer_offsets-23,__consumer_offsets-49,logaudit_20220314-8,__consumer_offsets-47,__consumer_offsets-16,__consumer_offsets-28,logaudit_20220307-1,logaudit_20220314-3,__consumer_offsets-31,__consumer_offsets-36,__consumer_offsets-42,__consumer_offsets-3,logaudit_20220307-7,__consumer_offsets-18,__consumer_offsets-37,__consumer_offsets-15,__consumer_offsets-24,logaudit_20220307-6,logaudit_20220314-9,logaudit_20220314-4,__consumer_offsets-38,__consumer_offsets-17,logaudit_20220307-9,__consumer_offsets-48,__consumer_offsets-19,logaudit_20220307-2,__consumer_offsets-11,__consumer_offsets-13,__consumer_offsets-2,__consumer_offsets-43,__consumer_offsets-6,__consumer_offsets-14,logaudit_20220314-5,logaudit_20220314-1,logaudit_20220307-5,logaudit_20220314-6,__consumer_offsets-20,__consumer_offsets-0,logaudit_20220314-10,__consumer_offsets-44,__consumer_offsets-39,logaudit_20220307-3,__consumer_offsets-12,yanbiao_1-0,logaudit_20220307-10,__consumer_offsets-45,__consumer_offsets-1,__consumer_offsets-5,__consumer_offsets-26,__consumer_offsets-29,__consumer_offsets-34,__consumer_offsets-10,__consumer_offsets-32,logaudit_20220307-0,__consumer_offsets-40 and stopped moving logs for partitions  because they are in the failed log directory /tmp/kafka-logs. (kafka.server.ReplicaManager)
[2022-03-28 10:36:39,279] INFO Stopping serving logs in dir /tmp/kafka-logs (kafka.log.LogManager)
[2022-03-28 10:36:39,633] ERROR Shutdown broker because all log dirs in /tmp/kafka-logs have failed (kafka.log.LogManager)

報錯位置的源碼：

private def cleanLog(cleanable: LogToClean): Unit = {
    val startOffset = cleanable.firstDirtyOffset
​    var endOffset = startOffset
​    try {
​        val (nextDirtyOffset, cleanerStats) = cleaner.clean(cleanable)
​        endOffset = nextDirtyOffset
​        recordStats(cleaner.id, cleanable.log.name, startOffset, endOffset, cleanerStats)
​      } catch {
​        case _: LogCleaningAbortedException => // task can be aborted, let it go.
​        case _: KafkaStorageException => // partition is already offline. let it go.
​        case e: IOException =>
​        val logDirectory = cleanable.log.parentDir
​        val msg = s"Failed to clean up log for ${cleanable.topicPartition} in dir $logDirectory due to IOException"
​        logDirFailureChannel.maybeAddOfflineLogDir(logDirectory, msg, e)
​      } finally {
​        cleanerManager.doneCleaning(cleanable.topicPartition, cleanable.log.parentDirFile, endOffset)
​      }
​    }

顯示是因為找不到文件報錯，我在前面安裝部署的時候log文件的位置是log.dirs=/tmp/kafka-logs，說明這里面的文件被悄悄的清理了

因為沒人會無緣無故到這里來手動清理，網上比較靠譜的解釋是linux系統會定時清理/tmp下的文件，解決該問題的方案也是先清空log.dirs下的文件，然后重啟broker

但是這個方案只是臨時的，並且暴力全部刪除會丟失數據（kafka消息就是通過log文件記錄的）或數據混亂，在kafka的官網issue里找到的問題單也做了說明，目前沒有解決此問題：

https://issues.apache.org/jira/browse/KAFKA-6188

2、關鍵字

java.lang.ClassNotFoundException: org.springframework.kafka.listener.ContainerProperties

org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'kafkaListenerContainerFactoryConfigurer' defined in class path resource [org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.class]: Post-processing of merged bean definition failed; nested exception is java.lang.IllegalStateException: Failed to introspect Class [org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer] from ClassLoader [sun.misc.Launcher$AppClassLoader@18b4aac2]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:569) ~[spring-beans-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:514) ~[spring-beans-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:321) ~[spring-beans-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:234) ~[spring-beans-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:319) ~[spring-beans-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:199) ~[spring-beans-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:866) ~[spring-beans-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:878) ~[spring-context-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:550) ~[spring-context-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:141) ~[spring-boot-2.1.18.RELEASE.jar:2.1.18.RELEASE]
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:744) ~[spring-boot-2.1.18.RELEASE.jar:2.1.18.RELEASE]
    at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:391) ~[spring-boot-2.1.18.RELEASE.jar:2.1.18.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:312) ~[spring-boot-2.1.18.RELEASE.jar:2.1.18.RELEASE]
    at com.yb.YbTestApplication.main(YbTestApplication.java:13) [classes/:na]
Caused by: java.lang.IllegalStateException: Failed to introspect Class [org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer] from ClassLoader [sun.misc.Launcher$AppClassLoader@18b4aac2]
    at org.springframework.util.ReflectionUtils.getDeclaredMethods(ReflectionUtils.java:507) ~[spring-core-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.util.ReflectionUtils.doWithLocalMethods(ReflectionUtils.java:367) ~[spring-core-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.beans.factory.annotation.InitDestroyAnnotationBeanPostProcessor.buildLifecycleMetadata(InitDestroyAnnotationBeanPostProcessor.java:207) ~[spring-beans-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.beans.factory.annotation.InitDestroyAnnotationBeanPostProcessor.findLifecycleMetadata(InitDestroyAnnotationBeanPostProcessor.java:189) ~[spring-beans-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.beans.factory.annotation.InitDestroyAnnotationBeanPostProcessor.postProcessMergedBeanDefinition(InitDestroyAnnotationBeanPostProcessor.java:128) ~[spring-beans-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.context.annotation.CommonAnnotationBeanPostProcessor.postProcessMergedBeanDefinition(CommonAnnotationBeanPostProcessor.java:298) ~[spring-context-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.applyMergedBeanDefinitionPostProcessors(AbstractAutowireCapableBeanFactory.java:1075) ~[spring-beans-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:566) ~[spring-beans-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    ... 13 common frames omitted
Caused by: java.lang.NoClassDefFoundError: org/springframework/kafka/listener/ContainerProperties
    at java.lang.Class.getDeclaredMethods0(Native Method) ~[na:1.8.0_181]
    at java.lang.Class.privateGetDeclaredMethods(Class.java:2701) ~[na:1.8.0_181]
    at java.lang.Class.getDeclaredMethods(Class.java:1975) ~[na:1.8.0_181]
    at org.springframework.util.ReflectionUtils.getDeclaredMethods(ReflectionUtils.java:489) ~[spring-core-5.1.19.RELEASE.jar:5.1.19.RELEASE]
    ... 20 common frames omitted
Caused by: java.lang.ClassNotFoundException: org.springframework.kafka.listener.ContainerProperties
    at java.net.URLClassLoader.findClass(URLClassLoader.java:381) ~[na:1.8.0_181]
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424) ~[na:1.8.0_181]
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349) ~[na:1.8.0_181]
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ~[na:1.8.0_181]
    ... 24 common frames omitted

原因：springboot項目，版本是2.5.13，集成的spring-kafka版本是自定義的2.1.14.RELEASE,使用了@KafkaListener 

ConcurrentKafkaListenerContainerFactoryConfigurer和ContainerProperties這兩個類和springboot默認版本的spring-kafka不符
解決方案：去掉自定義的版本（不指定），使用默認的springboot的版本