嵌入式kafka无法启动-错误

goucqfw6  于 2021-06-08  发布在  Kafka
关注(0)|答案(2)|浏览(664)

我很难解决这个问题。这是我的junit，我使用的是嵌入spring的Kafka。当我运行我的测试用例时，我遇到了奇怪的问题/异常。

  1. @RunWith(SpringRunner.class)
  2. @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
  3. @ActiveProfiles("test")
  4. @FixMethodOrder(MethodSorters.NAME_ASCENDING)
  5. @DirtiesContext
  6. public class WatchListUpdateTest {
  8.  @ClassRule
  9.  public static KafkaEmbedded KAFKA = new KafkaEmbedded(1, true, "abc");
  11.  @BeforeClass
  12.  public static void startKafka() throws Exception {
  13.   String kafkaBootstrapServers = KAFKA.getBrokersAsString();
  14.   System.out.print("[Embedded Kafka Server:{}]" + kafkaBootstrapServers);
  15.   System.setProperty("kafka.consumer.bootstrap.servers", kafkaBootstrapServers);
  16.   System.setProperty("kafka.producer.bootstrap.servers", kafkaBootstrapServers);
  18.  }
  20.  @Autowired
  21.  ApplicationContext applicationContext;
  23.  private KafkaTestHelper helper = new KafkaTestHelper(KAFKA, "abc");
  25.  @Before
  26.  public void setUp() throws Exception {
  27.   helper.start(KAFKA.getPartitionsPerTopic());
  28.  }
  30.  @After
  31.  public void tearDown() throws Exception {
  32.   helper.stop();
  33.  }
  35.  @Test
  36.  public void testIng() throws Exception {
  38.  }
  40. }

和

  1. import java.util.Map;
  2. import java.util.concurrent.BlockingQueue;
  3. import java.util.concurrent.LinkedBlockingQueue;
  5. import org.apache.kafka.clients.consumer.ConsumerRecord;
  6. import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
  7. import org.springframework.kafka.listener.KafkaMessageListenerContainer;
  8. import org.springframework.kafka.listener.MessageListener;
  9. import org.springframework.kafka.listener.config.ContainerProperties;
  10. import org.springframework.kafka.test.rule.KafkaEmbedded;
  11. import org.springframework.kafka.test.utils.ContainerTestUtils;
  12. import org.springframework.kafka.test.utils.KafkaTestUtils;
  14. public class KafkaTestHelper {
  16.  private KafkaMessageListenerContainer < String, String > container;
  18.  private BlockingQueue < ConsumerRecord < String, String >> records;
  20.  public KafkaTestHelper(KafkaEmbedded KAFKA, String topics) {
  22.   Map < String, Object > consumerProperties = KafkaTestUtils.consumerProps("sender", "false", KAFKA);
  24.   DefaultKafkaConsumerFactory < String, String > consumerFactory = new DefaultKafkaConsumerFactory < String, String > (
  25.    consumerProperties);
  27.   ContainerProperties containerProperties = new ContainerProperties(topics);
  29.   container = new KafkaMessageListenerContainer < > (consumerFactory, containerProperties);
  31.   records = new LinkedBlockingQueue < > ();
  33.   container.setupMessageListener(new MessageListener < String, String > () {
  34.    @Override
  35.    public void onMessage(ConsumerRecord < String, String > record) {
  36.     records.add(record);
  37.    }
  38.   });
  40.  }
  42.  public void start(int numberOfPartitions) {
  43.   container.start();
  44.   try {
  45.    ContainerTestUtils.waitForAssignment(container, numberOfPartitions);
  46.   } catch (Exception e) {
  47.    e.printStackTrace();
  48.   }
  50.  }
  52.  public void stop() {
  53.   container.stop();
  54.  }
  56. }

我有个例外：

  1. 01: 38: 17.454[main - SendThread(127.0 .0 .1: 64964)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid: 0x16494b8ced60001, packet::clientPath: null serverPath: null finished: false header::27, 1 replyHeader::27, 18, -101 request::'/cluster/id,#7b2276657273696f6e223a2231222c226964223a22635f686f3459694d547643637a6849386465436d7841227d,v{s{31,s{'
  2. world, 'anyone}}},0  response::  
  3. 01: 38: 17.455[SyncThread: 0] DEBUG org.apache.zookeeper.server.FinalRequestProcessor - Processing request::sessionid: 0x16494b8ced60001 type: create cxid: 0x1c zxid: 0x13 txntype: 1 reqpath: n / a
  4. 01: 38: 17.455[SyncThread: 0] DEBUG org.apache.zookeeper.server.FinalRequestProcessor - sessionid: 0x16494b8ced60001 type: create cxid: 0x1c zxid: 0x13 txntype: 1 reqpath: n / a
  5. 01: 38: 17.455[main - SendThread(127.0 .0 .1: 64964)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid: 0x16494b8ced60001, packet::clientPath: null serverPath: null finished: false header::28, 1 replyHeader::28, 19, 0 request::'/cluster,,v{s{31,s{'
  6. world, 'anyone}}},0  response:: ' / cluster
  7. 01: 38: 17.456[SyncThread: 0] DEBUG org.apache.zookeeper.server.FinalRequestProcessor - Processing request::sessionid: 0x16494b8ced60001 type: create cxid: 0x1d zxid: 0x14 txntype: 1 reqpath: n / a
  8. 01: 38: 17.456[SyncThread: 0] DEBUG org.apache.zookeeper.server.FinalRequestProcessor - sessionid: 0x16494b8ced60001 type: create cxid: 0x1d zxid: 0x14 txntype: 1 reqpath: n / a
  9. 01: 38: 17.456[main - SendThread(127.0 .0 .1: 64964)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid: 0x16494b8ced60001, packet::clientPath: null serverPath: null finished: false header::29, 1 replyHeader::29, 20, 0 request::'/cluster/id,#7b2276657273696f6e223a2231222c226964223a22635f686f3459694d547643637a6849386465436d7841227d,v{s{31,s{'
  10. world, 'anyone}}},0  response:: ' / cluster / id
  11. 01: 38: 17.457[main] INFO kafka.server.KafkaServer - Cluster ID = c_ho4YiMTvCczhI8deCmxA
  12. 01: 38: 17.460[main] WARN kafka.server.BrokerMetadataCheckpoint - No meta.properties file under dir /
  13.  var / folders / _s / k06t9c8x7470lcccm39r96fc0000gp / T / kafka - 72082059269987836 / meta.properties
  14. 01: 38: 17.482[ThrottledRequestReaper - Fetch] INFO kafka.server.ClientQuotaManager$ThrottledRequestReaper - [ThrottledRequestReaper - Fetch], Starting
  15. 01: 38: 17.483[main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name Fetch - delayQueue
  16. 01: 38: 17.484[main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name Produce - delayQueue
  17. 01: 38: 17.484[ThrottledRequestReaper - Produce] INFO kafka.server.ClientQuotaManager$ThrottledRequestReaper - [ThrottledRequestReaper - Produce], Starting
  18. 01: 38: 17.519[SyncThread: 0] DEBUG org.apache.zookeeper.server.FinalRequestProcessor - Processing request::sessionid: 0x16494b8ced60001 type: getChildren cxid: 0x1e zxid: 0xfffffffffffffffe txntype: unknown reqpath: /brokers/topics
  19. 01: 38: 17.519[SyncThread: 0] DEBUG org.apache.zookeeper.server.FinalRequestProcessor - sessionid: 0x16494b8ced60001 type: getChildren cxid: 0x1e zxid: 0xfffffffffffffffe txntype: unknown reqpath: /brokers/topics
  20. 01: 38: 17.520[main - SendThread(127.0 .0 .1: 64964)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid: 0x16494b8ced60001, packet::clientPath: null serverPath: null finished: false header::30, 8 replyHeader::30, 20, 0 request::'/brokers/topics,F  response:: v{} 
  21. 01: 38: 17.540[main] INFO kafka.log.LogManager - Loading logs.
  22. 01: 38: 17.546[main] INFO kafka.log.LogManager - Logs loading complete in 6 ms.
  23. 01: 38: 17.567[main] INFO kafka.log.LogManager - Starting log cleanup with a period of 300000 ms.
  24. 01: 38: 17.568[main] DEBUG kafka.utils.KafkaScheduler - Scheduling task kafka - log - retention with initial delay 30000 ms and period 300000 ms.
  25. 01: 38: 17.569[main] INFO kafka.log.LogManager - Starting log flusher with a
  26. default period of 9223372036854775807 ms.
  27. 01: 38: 17.569[main] DEBUG kafka.utils.KafkaScheduler - Scheduling task kafka - log - flusher with initial delay 30000 ms and period 9223372036854775807 ms.
  28. 01: 38: 17.570[main] DEBUG kafka.utils.KafkaScheduler - Scheduling task kafka - recovery - point - checkpoint with initial delay 30000 ms and period 60000 ms.
  29. 01: 38: 17.570[main] DEBUG kafka.utils.KafkaScheduler - Scheduling task kafka - delete - logs with initial delay 30000 ms and period 60000 ms.
  30. 01: 38: 17.571[main] INFO kafka.log.LogCleaner - Starting the log cleaner
  31. 01: 38: 17.572[kafka - log - cleaner - thread - 0] INFO kafka.log.LogCleaner - [kafka - log - cleaner - thread - 0], Starting
  32. 01: 38: 17.600[main] ERROR kafka.server.KafkaServer - [Kafka Server 0], Fatal error during KafkaServer startup.Prepare to shutdown
  33. java.lang.NoSuchMethodError: org.apache.kafka.common.network.ChannelBuilders.serverChannelBuilder(Lorg / apache / kafka / common / protocol / SecurityProtocol; Ljava / util / Map; Lorg / apache / kafka / common / security / authenticator / CredentialCache;) Lorg / apache / kafka / common / network / ChannelBuilder;
  34. at kafka.network.Processor. < init > (SocketServer.scala: 422)
  35. at kafka.network.SocketServer.newProcessor(SocketServer.scala: 145)
  36. at kafka.network.SocketServer$$anonfun$startup$1$$anonfun$apply$1.apply$mcVI$sp(SocketServer.scala: 96)
  37. at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala: 160)
  38. at kafka.network.SocketServer$$anonfun$startup$1.apply(SocketServer.scala: 95)
  39. at kafka.network.SocketServer$$anonfun$startup$1.apply(SocketServer.scala: 90)
  40. at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala: 59)
  41. at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala: 48)
  42. at kafka.network.SocketServer.startup(SocketServer.scala: 90)
  43. at kafka.server.KafkaServer.startup(KafkaServer.scala: 215)
  44. at kafka.utils.TestUtils$.createServer(TestUtils.scala: 124)
  45. at kafka.utils.TestUtils.createServer(TestUtils.scala)
  46. at org.springframework.kafka.test.rule.KafkaEmbedded.before(KafkaEmbedded.java: 156)
  47. at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java: 46)
  48. at org.junit.rules.RunRules.evaluate(RunRules.java: 20)
  49. at org.junit.runners.ParentRunner.run(ParentRunner.java: 363)
  50. at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java: 191)
  51. at org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java: 86)
  52. at org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java: 38)
  53. at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java: 538)
  54. at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java: 760)
  55. at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java: 460)
  56. at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java: 206)
  57. 01: 38: 17.601[main] INFO kafka.server.KafkaServer - [Kafka Server 0], shutting down
  58. 01: 38: 17.603[main] INFO kafka.network.SocketServer - [Socket Server on Broker 0], Shutting down
  59. 01: 38: 17.604[main] WARN kafka.utils.CoreUtils$ - null
  60. java.lang.NullPointerException: null
  61. at kafka.network.SocketServer$$anonfun$shutdown$3.apply(SocketServer.scala: 129)
  62. at kafka.network.SocketServer$$anonfun$shutdown$3.apply(SocketServer.scala: 129)
  63. at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala: 33)
  64. at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala: 186)
  65. at kafka.network.SocketServer.shutdown(SocketServer.scala: 129)
  66. at kafka.server.KafkaServer$$anonfun$shutdown$2.apply$mcV$sp(KafkaServer.scala: 582)
  67. at kafka.utils.CoreUtils$.swallow(CoreUtils.scala: 78)
  68. at kafka.utils.Logging$class.swallowWarn(Logging.scala: 94)
  69. at kafka.utils.CoreUtils$.swallowWarn(CoreUtils.scala: 48)
  70. at kafka.utils.Logging$class.swallow(Logging.scala: 96)
  71. at kafka.utils.CoreUtils$.swallow(CoreUtils.scala: 48)
  72. at kafka.server.KafkaServer.shutdown(KafkaServer.scala: 582)
  73. at kafka.server.KafkaServer.startup(KafkaServer.scala: 289)
  74. at kafka.utils.TestUtils$.createServer(TestUtils.scala: 124)
  75. at kafka.utils.TestUtils.createServer(TestUtils.scala)
  76. at org.springframework.kafka.test.rule.KafkaEmbedded.before(KafkaEmbedded.java: 156)
  77. at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java: 46)
  78. at org.junit.rules.RunRules.evaluate(RunRules.java: 20)
  79. at org.junit.runners.ParentRunner.run(ParentRunner.java: 363)
  80. at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java: 191)
  81. at org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java: 86)
  82. at org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java: 38)
  83. at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java: 538)
  84. at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java: 760)
  85. at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java: 460)
  86. at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java: 206)
  87. 01: 38: 17.605[main] DEBUG kafka.utils.KafkaScheduler - Shutting down task scheduler.
  88. 01: 38: 17.606[main] INFO kafka.log.LogManager - Shutting down.
  89. 01: 38: 17.607[main] INFO kafka.log.LogCleaner - Shutting down the log cleaner.
  90. 01: 38: 17.608[main] INFO kafka.log.LogCleaner - [kafka - log - cleaner - thread - 0], Shutting down
  91. 01: 38: 17.608[kafka - log - cleaner - thread - 0] INFO kafka.log.LogCleaner - [kafka - log - cleaner - thread - 0], Stopped
  92. 01: 38: 17.608[main] INFO kafka.log.LogCleaner - [kafka - log - cleaner - thread - 0], Shutdown completed

我使用的是springboot1.5.4.release和kafka 0.11.0.0

  1. <dependency>
  2.             <groupId>org.springframework.kafka</groupId>
  3.             <artifactId>spring-kafka-test</artifactId>
  4.             <version>1.5.4.RELEASE}</version>
  5.             <scope>test</scope>
  6. </dependency>   
  8. <dependency>
  9.             <groupId>org.apache.kafka</groupId>
  10.             <artifactId>kafka-clients</artifactId>
  11.             <version>0.11.0.0</version>
  12. </dependency>

请帮我解决这个问题，我已经花了很多时间，但无法解决这个问题。提前谢谢。

apache-kafkaspring-bootjunitembedded-kafka

2条答案

按热度按时间
cyvaqqii

cyvaqqii1#

这是的依赖项集 Kafka 2.4.1 ，这可能会解决您的问题（来源：https://docs.spring.io/spring-kafka/reference/html/#update-deps）：哪里 {kafka-version} 是 2.4.1 以及 {springKafkaTest} 是 2.4.8.RELEASE ```

org.springframework.kafka
spring-kafka-test
{springKafkaTest}
test

赞(0）分享  回复(0）举报  2021-06-08
ruyhziif

ruyhziif2#

在加上这种依赖性之后，它对我起了作用

  1. <dependency>
  2.         <groupId>org.apache.kafka</groupId>
  3.         <artifactId>kafka_2.11</artifactId>
  4.         <version>${kafka.version}</version>
  5.         <scope>test</scope>
  6.         <exclusions>
  7.             <exclusion>
  8.                 <groupId>org.slf4j</groupId>
  9.                 <artifactId>slf4j-log4j12</artifactId>
  10.             </exclusion>
  11.             <exclusion>
  12.                 <groupId>org.apache.kafka</groupId>
  13.                 <artifactId>kafka-clients</artifactId>
  14.             </exclusion>
  15.         </exclusions>
  16.     </dependency>
展开查看全部
赞(0）分享  回复(0）举报  2021-06-08
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com