spring-boot - 嵌入式kafka无法启动-错误

Question

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

@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
@ActiveProfiles("test")
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
@DirtiesContext
public class WatchListUpdateTest {

 @ClassRule
 public static KafkaEmbedded KAFKA = new KafkaEmbedded(1, true, "abc");

 @BeforeClass
 public static void startKafka() throws Exception {
  String kafkaBootstrapServers = KAFKA.getBrokersAsString();
  System.out.print("[Embedded Kafka Server:{}]" + kafkaBootstrapServers);
  System.setProperty("kafka.consumer.bootstrap.servers", kafkaBootstrapServers);
  System.setProperty("kafka.producer.bootstrap.servers", kafkaBootstrapServers);

 }

 @Autowired
 ApplicationContext applicationContext;

 private KafkaTestHelper helper = new KafkaTestHelper(KAFKA, "abc");

 @Before
 public void setUp() throws Exception {
  helper.start(KAFKA.getPartitionsPerTopic());
 }

 @After
 public void tearDown() throws Exception {
  helper.stop();
 }

 @Test
 public void testIng() throws Exception {

 }

} 

和

import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.ContainerTestUtils;
import org.springframework.kafka.test.utils.KafkaTestUtils;

public class KafkaTestHelper {

 private KafkaMessageListenerContainer < String, String > container;

 private BlockingQueue < ConsumerRecord < String, String >> records;

 public KafkaTestHelper(KafkaEmbedded KAFKA, String topics) {

  Map < String, Object > consumerProperties = KafkaTestUtils.consumerProps("sender", "false", KAFKA);

  DefaultKafkaConsumerFactory < String, String > consumerFactory = new DefaultKafkaConsumerFactory < String, String > (
   consumerProperties);

  ContainerProperties containerProperties = new ContainerProperties(topics);

  container = new KafkaMessageListenerContainer < > (consumerFactory, containerProperties);

  records = new LinkedBlockingQueue < > ();

  container.setupMessageListener(new MessageListener < String, String > () {
   @Override
   public void onMessage(ConsumerRecord < String, String > record) {
    records.add(record);
   }
  });


 }

 public void start(int numberOfPartitions) {
  container.start();
  try {
   ContainerTestUtils.waitForAssignment(container, numberOfPartitions);
  } catch (Exception e) {
   e.printStackTrace();
  }

 }

 public void stop() {
  container.stop();
 }

}

这是我的异常(exception):

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{'
world, 'anyone}}},0  response::  
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
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
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{'
world, 'anyone}}},0  response:: ' / cluster
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
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
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{'
world, 'anyone}}},0  response:: ' / cluster / id
01: 38: 17.457[main] INFO kafka.server.KafkaServer - Cluster ID = c_ho4YiMTvCczhI8deCmxA
01: 38: 17.460[main] WARN kafka.server.BrokerMetadataCheckpoint - No meta.properties file under dir /
 var / folders / _s / k06t9c8x7470lcccm39r96fc0000gp / T / kafka - 72082059269987836 / meta.properties
01: 38: 17.482[ThrottledRequestReaper - Fetch] INFO kafka.server.ClientQuotaManager$ThrottledRequestReaper - [ThrottledRequestReaper - Fetch], Starting
01: 38: 17.483[main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name Fetch - delayQueue
01: 38: 17.484[main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name Produce - delayQueue
01: 38: 17.484[ThrottledRequestReaper - Produce] INFO kafka.server.ClientQuotaManager$ThrottledRequestReaper - [ThrottledRequestReaper - Produce], Starting
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
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
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{} 
01: 38: 17.540[main] INFO kafka.log.LogManager - Loading logs.
01: 38: 17.546[main] INFO kafka.log.LogManager - Logs loading complete in 6 ms.
01: 38: 17.567[main] INFO kafka.log.LogManager - Starting log cleanup with a period of 300000 ms.
01: 38: 17.568[main] DEBUG kafka.utils.KafkaScheduler - Scheduling task kafka - log - retention with initial delay 30000 ms and period 300000 ms.
01: 38: 17.569[main] INFO kafka.log.LogManager - Starting log flusher with a
default period of 9223372036854775807 ms.
01: 38: 17.569[main] DEBUG kafka.utils.KafkaScheduler - Scheduling task kafka - log - flusher with initial delay 30000 ms and period 9223372036854775807 ms.
01: 38: 17.570[main] DEBUG kafka.utils.KafkaScheduler - Scheduling task kafka - recovery - point - checkpoint with initial delay 30000 ms and period 60000 ms.
01: 38: 17.570[main] DEBUG kafka.utils.KafkaScheduler - Scheduling task kafka - delete - logs with initial delay 30000 ms and period 60000 ms.
01: 38: 17.571[main] INFO kafka.log.LogCleaner - Starting the log cleaner
01: 38: 17.572[kafka - log - cleaner - thread - 0] INFO kafka.log.LogCleaner - [kafka - log - cleaner - thread - 0], Starting
01: 38: 17.600[main] ERROR kafka.server.KafkaServer - [Kafka Server 0], Fatal error during KafkaServer startup.Prepare to shutdown
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;
at kafka.network.Processor. < init > (SocketServer.scala: 422)
at kafka.network.SocketServer.newProcessor(SocketServer.scala: 145)
at kafka.network.SocketServer$$anonfun$startup$1$$anonfun$apply$1.apply$mcVI$sp(SocketServer.scala: 96)
at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala: 160)
at kafka.network.SocketServer$$anonfun$startup$1.apply(SocketServer.scala: 95)
at kafka.network.SocketServer$$anonfun$startup$1.apply(SocketServer.scala: 90)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala: 59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala: 48)
at kafka.network.SocketServer.startup(SocketServer.scala: 90)
at kafka.server.KafkaServer.startup(KafkaServer.scala: 215)
at kafka.utils.TestUtils$.createServer(TestUtils.scala: 124)
at kafka.utils.TestUtils.createServer(TestUtils.scala)
at org.springframework.kafka.test.rule.KafkaEmbedded.before(KafkaEmbedded.java: 156)
at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java: 46)
at org.junit.rules.RunRules.evaluate(RunRules.java: 20)
at org.junit.runners.ParentRunner.run(ParentRunner.java: 363)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java: 191)
at org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java: 86)
at org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java: 38)
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java: 538)
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java: 760)
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java: 460)
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java: 206)
01: 38: 17.601[main] INFO kafka.server.KafkaServer - [Kafka Server 0], shutting down
01: 38: 17.603[main] INFO kafka.network.SocketServer - [Socket Server on Broker 0], Shutting down
01: 38: 17.604[main] WARN kafka.utils.CoreUtils$ - null
java.lang.NullPointerException: null
at kafka.network.SocketServer$$anonfun$shutdown$3.apply(SocketServer.scala: 129)
at kafka.network.SocketServer$$anonfun$shutdown$3.apply(SocketServer.scala: 129)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala: 33)
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala: 186)
at kafka.network.SocketServer.shutdown(SocketServer.scala: 129)
at kafka.server.KafkaServer$$anonfun$shutdown$2.apply$mcV$sp(KafkaServer.scala: 582)
at kafka.utils.CoreUtils$.swallow(CoreUtils.scala: 78)
at kafka.utils.Logging$class.swallowWarn(Logging.scala: 94)
at kafka.utils.CoreUtils$.swallowWarn(CoreUtils.scala: 48)
at kafka.utils.Logging$class.swallow(Logging.scala: 96)
at kafka.utils.CoreUtils$.swallow(CoreUtils.scala: 48)
at kafka.server.KafkaServer.shutdown(KafkaServer.scala: 582)
at kafka.server.KafkaServer.startup(KafkaServer.scala: 289)
at kafka.utils.TestUtils$.createServer(TestUtils.scala: 124)
at kafka.utils.TestUtils.createServer(TestUtils.scala)
at org.springframework.kafka.test.rule.KafkaEmbedded.before(KafkaEmbedded.java: 156)
at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java: 46)
at org.junit.rules.RunRules.evaluate(RunRules.java: 20)
at org.junit.runners.ParentRunner.run(ParentRunner.java: 363)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java: 191)
at org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java: 86)
at org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java: 38)
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java: 538)
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java: 760)
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java: 460)
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java: 206)
01: 38: 17.605[main] DEBUG kafka.utils.KafkaScheduler - Shutting down task scheduler.
01: 38: 17.606[main] INFO kafka.log.LogManager - Shutting down.
01: 38: 17.607[main] INFO kafka.log.LogCleaner - Shutting down the log cleaner.
01: 38: 17.608[main] INFO kafka.log.LogCleaner - [kafka - log - cleaner - thread - 0], Shutting down
01: 38: 17.608[kafka - log - cleaner - thread - 0] INFO kafka.log.LogCleaner - [kafka - log - cleaner - thread - 0], Stopped
01: 38: 17.608[main] INFO kafka.log.LogCleaner - [kafka - log - cleaner - thread - 0], Shutdown completed

我正在使用 spring boot 1.5.4.RELEASE & kafka 0.11.0.0

<dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <version>1.5.4.RELEASE}</version>
            <scope>test</scope>
</dependency>   

<dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.11.0.0</version>
</dependency>

请帮我解决这个问题，我花了很多时间但无法弄清楚这个问题。提前致谢。

Answer 1

添加这个依赖后它对我有用

    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.11</artifactId>
        <version>${kafka.version}</version>
        <scope>test</scope>
        <exclusions>
            <exclusion>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
            </exclusion>
            <exclusion>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
            </exclusion>
        </exclusions>
    </dependency>