kafka使用过程中的坑及解决方法

布满荆棘的人生 2023-10-07 22:41 20阅读 0赞

#### 文章目录 ####

*   *  Invalid value org.apache.kafka.common.serialization.StringSerializer for configuration key.serializer: Class org.apache.kafka.common.serialization.StringSerializer could not be found.
     *   *  场景描述
         *  报错日志
         *  解决方案

### Invalid value org.apache.kafka.common.serialization.StringSerializer for configuration key.serializer: Class org.apache.kafka.common.serialization.StringSerializer could not be found. ###

#### 场景描述 ####

springboot中使用kafka多线程发送消息，在本地启动时能正常发送消息，但是当服务部署到linux服务器后，一直报异常，消息发送失败。

**修改后的demo代码**

KafkaProducer kafkaProducer = new KafkaProducer(kafkaProducerProperties);
    //开启15个线程
    ForkJoinPool pool = new ForkJoinPool(15);
    Future future = pool.submit(() -> sendList.parallelStream().forEach(dto -> {
    	log.info("thread:" + Thread.currentThread().getName()
    			+ "参数" + dto);
    	//多线程发送时一定要加这句话，否则会报异常 org.apache.kafka.common.config.ConfigException: Invalid value org.apache.kafka.common.serialization.StringSerializer for configuration key.serializer: Class org.apache.kafka.common.serialization.StringSerializer could not be found.
    	Thread.currentThread().setContextClassLoader(null);
    	kafkaProducer.sendMessageSync("test-topic", jsonObject.toJSONString());
    }));
    
    try {
        
    	future.get();
    } catch (InterruptedException e) {
        
    	log.error("kafka发送数据异常：", e);
    	Thread.currentThread().interrupt();
    } catch (ExecutionException e) {
        
    	log.error("kafka发送数据异常：", e);
    	Thread.currentThread().interrupt();
    } finally {
        
    	pool.shutdownNow();
    }

#### 报错日志 ####

org.apache.kafka.common.config.ConfigException: Invalid value org.apache.kafka.common.serialization.StringSerializer for configuration key.serializer: Class org.apache.kafka.common.serialization.StringSerializer could not be found.
    	at org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:744)
    	at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:490)
    	at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:483)
    	at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:108)
    	at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:129)
    	at org.apache.kafka.clients.producer.ProducerConfig.<init>(ProducerConfig.java:536)
    	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:291)
    	at org.springframework.kafka.core.DefaultKafkaProducerFactory.createRawProducer(DefaultKafkaProducerFactory.java:668)
    	at org.springframework.kafka.core.DefaultKafkaProducerFactory.createKafkaProducer(DefaultKafkaProducerFactory.java:542)
    	at org.springframework.kafka.core.DefaultKafkaProducerFactory.doCreateProducer(DefaultKafkaProducerFactory.java:519)
    	at org.springframework.kafka.core.DefaultKafkaProducerFactory.createProducer(DefaultKafkaProducerFactory.java:480)
    	at org.springframework.kafka.core.DefaultKafkaProducerFactory.createProducer(DefaultKafkaProducerFactory.java:474)
    	at org.springframework.kafka.core.KafkaTemplate.getTheProducer(KafkaTemplate.java:665)
    	at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:551)
    	at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:363)
    	at com.test.common.kafka.producer.KafkaProducer.sendMessageSync(KafkaProducer.java:164)
    	at com.test.demo.tools.service.impl.KafkaServiceImpl.threadSendMessage(KafkaServiceImpl.java:103)
    	at com.test.demo.tools.service.impl.KafkaServiceImpl.lambda$threadSendMessage$0(KafkaServiceImpl.java:81)
    	at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
    	at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1655)
    	at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
    	at java.base/java.util.stream.ForEachOps$ForEachTask.compute(ForEachOps.java:290)
    	at java.base/java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:746)
    	at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
    	at java.base/java.util.concurrent.ForkJoinTask.doInvoke(ForkJoinTask.java:408)
    	at java.base/java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:736)
    	at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateParallel(ForEachOps.java:159)
    	at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateParallel(ForEachOps.java:173)
    	at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233)
    	at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:497)
    	at java.base/java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:661)
    	at com.test.demo.tools.service.impl.KafkaServiceImpl.lambda$threadSendMessage$1(KafkaServiceImpl.java:78)
    	at java.base/java.util.concurrent.ForkJoinTask$AdaptedRunnableAction.exec(ForkJoinTask.java:1407)
    	at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
    	at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
    	at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
    	at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
    	at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)

#### 解决方案 ####

是因为classloader不是同一个所以加载不到

在之前加上

Thread.currentThread().setContextClassLoader(null);

kafka client use Class.forName(trimmed, true, Utils.getContextOrKafkaClassLoader()) to get the Class object, and the create the instance, the key point is the classLoader, which is specified by the last param, the implementation of method Utils.getContextOrKafkaClassLoader() is

public static ClassLoader getContextOrKafkaClassLoader() {
        
        ClassLoader cl = Thread.currentThread().getContextClassLoader();
        if (cl == null)
            return getKafkaClassLoader();
        else
            return cl;
    }

so, by default, the Class object of org.apache.kafka.common.serialization.StringSerializer is load by the applicationClassLoader, if your target class is not loaded by the applicationClassLoader, this problem will happend !

to solve the problem, simply set the ContextClassLoader of current thread to null before new KafkaProducer instance like this

Thread.currentThread().setContextClassLoader(null);
    Producer<String, String> producer = new KafkaProducer(props);