Spring Kafka-ConfigException：配置 key.subject.name.strategy 的无效值 TopicNameStrategy：找不到类 TopicNameStrategy答案

【问题标题】：Spring Kafka-ConfigException: Invalid value TopicNameStrategy for configuration key.subject.name.strategy: Class TopicNameStrategy could not be foundSpring Kafka-ConfigException：配置 key.subject.name.strategy 的无效值 TopicNameStrategy：找不到类 TopicNameStrategy
【发布时间】：2022-07-14 18:01:35
【问题描述】：

我们刚刚部署了一个 kafka 生产者来生产，并面临一个在非生产中没有弹出的奇怪问题。该服务是一个 Spring Boot 微服务，它接收 REST HTTP 请求并使用 spring kafka 将事件发布到主题上。微服务托管在 AWS ECS 上。此 API 的 java 版本设置为 java 11。以下是错误：

org.apache.kafka.common.KafkaException: Failed to construct kafka producer
    at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:441)
    at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:291)
    at org.springframework.kafka.core.DefaultKafkaProducerFactory.createRawProducer(DefaultKafkaProducerFactory.java:743)
    at org.springframework.kafka.core.DefaultKafkaProducerFactory.createKafkaProducer(DefaultKafkaProducerFactory.java:584)
    at org.springframework.kafka.core.DefaultKafkaProducerFactory.doCreateProducer(DefaultKafkaProducerFactory.java:544)
    at org.springframework.kafka.core.DefaultKafkaProducerFactory.createProducer(DefaultKafkaProducerFactory.java:519)
    at org.springframework.kafka.core.DefaultKafkaProducerFactory.createProducer(DefaultKafkaProducerFactory.java:513)
    at org.springframework.kafka.core.KafkaTemplate.getTheProducer(KafkaTemplate.java:683)
    at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:569)
    at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:386)
    at com.nab.ms.hs.lodgement.producer.HardshipCaseSubmitEventProducer.publishHardshipCaseSubmitEvent(HardshipCaseSubmitEventProducer.java:47)
    at com.nab.ms.hs.lodgement.application.CreateHardshipRequestService.lambda$publishHardshipCaseSubmitEvent$0(CreateHardshipRequestService.java:108)
    at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
    at com.nab.ms.hs.lodgement.application.CreateHardshipRequestService.processAccountRequest(CreateHardshipRequestService.java:103)
    at com.nab.ms.hs.lodgement.application.CreateHardshipRequestService.processNewHardshipRequest(CreateHardshipRequestService.java:75)
    at com.nab.ms.hs.lodgement.application.HardshipNewRequestService.lambda$processNewHardshipRequest$0(HardshipNewRequestService.java:46)
    at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1736)
    at java.base/java.util.concurrent.CompletableFuture$AsyncRun.exec(CompletableFuture.java:1728)
    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)
Caused by: java.lang.ExceptionInInitializerError: null
    at io.confluent.kafka.serializers.KafkaAvroSerializer.configure(KafkaAvroSerializer.java:50)
    at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:368)
    ... 22 common frames omitted
Caused by: org.apache.kafka.common.config.ConfigException: Invalid value io.confluent.kafka.serializers.subject.TopicNameStrategy for configuration key.subject.name.strategy: Class io.confluent.kafka.serializers.subject.TopicNameStrategy could not be found.
    at org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:729)

以下是我运行./gradlew dependencies --configuration=runtimeClasspath 命令时的依赖项

+--- org.springframework.kafka:spring-kafka:2.7.1
|    |    +--- org.springframework:spring-context:5.3.7 (*)
|    |    +--- org.springframework:spring-messaging:5.3.7
|    |    |    +--- org.springframework:spring-beans:5.3.7 (*)
|    |    |    \--- org.springframework:spring-core:5.3.7 (*)
|    |    +--- org.springframework:spring-tx:5.3.7 (*)
|    |    +--- org.springframework.retry:spring-retry:1.3.1
|    |    +--- org.apache.kafka:kafka-clients:2.7.1
|    |    |    +--- com.github.luben:zstd-jni:1.4.5-6
|    |    |    +--- org.lz4:lz4-java:1.7.1
|    |    |    +--- org.xerial.snappy:snappy-java:1.1.7.7
|    |    |    \--- org.slf4j:slf4j-api:1.7.30
|    |    +--- org.jetbrains.kotlin:kotlin-stdlib:1.4.32
|    |    |    +--- org.jetbrains.kotlin:kotlin-stdlib-common:1.4.32
|    |    |    \--- org.jetbrains:annotations:13.0
|    |    \--- com.google.code.findbugs:jsr305:3.0.2
|    +--- org.apache.avro:avro:1.10.2
|    |    +--- com.fasterxml.jackson.core:jackson-core:2.12.2 -> 2.11.4
|    |    +--- com.fasterxml.jackson.core:jackson-databind:2.12.2 -> 2.11.4 (*)
|    |    +--- org.apache.commons:commons-compress:1.20
|    |    \--- org.slf4j:slf4j-api:1.7.30
|    +--- org.apache.kafka:kafka-clients:2.7.1 (*)
|    +--- org.apache.kafka:kafka-streams:2.7.1 -> 2.6.2
|    |    +--- org.apache.kafka:kafka-clients:2.6.2 -> 2.7.1 (*)
|    |    +--- org.apache.kafka:connect-json:2.6.2
|    |    |    +--- org.apache.kafka:connect-api:2.6.2
|    |    |    |    +--- org.apache.kafka:kafka-clients:2.6.2 -> 2.7.1 (*)
|    |    |    |    \--- org.slf4j:slf4j-api:1.7.30
|    |    |    +--- com.fasterxml.jackson.core:jackson-databind:2.10.5.1 -> 2.11.4 (*)
|    |    |    +--- com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.10.5 -> 2.11.4 (*)
|    |    |    \--- org.slf4j:slf4j-api:1.7.30
|    |    +--- org.slf4j:slf4j-api:1.7.30
|    |    \--- org.rocksdb:rocksdbjni:5.18.4
|    +--- io.confluent:common-config:6.1.1
|    |    +--- io.confluent:common-utils:6.1.1
|    |    |    \--- org.slf4j:slf4j-api:1.7.30
|    |    \--- org.slf4j:slf4j-api:1.7.30
|    +--- io.confluent:common-utils:6.1.1 (*)
|    +--- io.confluent:kafka-avro-serializer:6.1.1
|    |    +--- org.apache.avro:avro:1.9.2 -> 1.10.2 (*)
|    |    +--- io.confluent:kafka-schema-serializer:6.1.1
|    |    |    +--- io.confluent:kafka-schema-registry-client:6.1.1
|    |    |    |    +--- org.apache.kafka:kafka-clients:6.1.1-ccs -> 2.7.1 (*)
|    |    |    |    +--- org.apache.avro:avro:1.9.2 -> 1.10.2 (*)
|    |    |    |    +--- com.fasterxml.jackson.core:jackson-databind:2.10.5.1 -> 2.11.4 (*)
|    |    |    |    +--- jakarta.ws.rs:jakarta.ws.rs-api:2.1.6
|    |    |    |    +--- org.glassfish.jersey.core:jersey-common:2.31 -> 2.32
|    |    |    |    |    +--- jakarta.ws.rs:jakarta.ws.rs-api:2.1.6
|    |    |    |    |    +--- jakarta.annotation:jakarta.annotation-api:1.3.5
|    |    |    |    |    +--- org.glassfish.hk2.external:jakarta.inject:2.6.1
|    |    |    |    |    \--- org.glassfish.hk2:osgi-resource-locator:1.0.3
|    |    |    |    +--- io.swagger:swagger-annotations:1.6.2
|    |    |    |    \--- io.confluent:common-utils:6.1.1 (*)
|    |    |    \--- io.confluent:common-utils:6.1.1 (*)
|    |    +--- io.confluent:kafka-schema-registry-client:6.1.1 (*)
|    |    \--- io.confluent:common-utils:6.1.1 (*)
|    +--- io.confluent:kafka-schema-registry-client:6.1.1 (*)
|    +--- io.confluent:kafka-streams-avro-serde:6.1.1
|    |    +--- io.confluent:kafka-avro-serializer:6.1.1 (*)
|    |    +--- io.confluent:kafka-schema-registry-client:6.1.1 (*)
|    |    +--- org.apache.avro:avro:1.9.2 -> 1.10.2 (*)
|    |    \--- io.confluent:common-utils:6.1.1 (*)
|    +--- org.springframework.boot:spring-boot-starter:2.4.5 -> 2.4.6 (*)

以下是生产者代码：

@Service
@Slf4j
public class EventProducer {

  private final KafkaTemplate<EventKey, SubmitEvent> hardshipProducer;
  
  @Value("${app.kafka.topic.hardship.case.submit.event.name}")
  private String topicName;

  public EventProducer(KafkaTemplate<EventKey, SubmitEvent> hardshipProducer) {
    this.hardshipProducer = hardshipProducer;
  }

  @SneakyThrows
  public void publishHardshipCaseSubmitEvent(HardshipCaseSubmitEvent hardshipCaseSubmitEvent, HardshipData hardshipData) {
    
    ListenableFuture<SendResult<EventKey, SubmitEvent>> future = hardshipProducer.send(topicName,
        EventKey.newBuilder().setCaseId(hardshipData.getHsCaseId()).build(),
        hardshipCaseSubmitEvent);
    future.addCallback(new ListenableFutureCallback<>() {
      @SneakyThrows
      @Override
      public void onFailure(@NonNull Throwable ex) {
        log.error("Exception = " + ex.getMessage() + " publishing hardshipCaseSubmitEvent for meId = " + hardshipCaseSubmitEvent.getData().getMeId() + ", correlation id=" + correlationId + ", caseId=" + hardshipData.getHsCaseId(), ex);
      }

      @Override
      public void onSuccess(SendResult<HardshipEventKey, HardshipCaseSubmitEvent> result) {
        
          log.info("hardshipCaseSubmitEvent event status = success, partition= {}, offset= {}, meId={}, correlation id={}, caseId={}",
              result.getRecordMetadata().partition(),
              result.getRecordMetadata().offset(),
              hardshipCaseSubmitEvent.getData().getMeId().toString(),
              correlationId,
              hardshipData.getHsCaseId());
        }
      
    });
    hardshipProducer.flush();
  }
}

还要注意，事件有时会成功生成，有时会因上述错误而失败。我已经记录了事件主体进行比较，发现没有任何区别。我检查了容器实例中存在的 war 文件，发现所有 jar 文件依赖项都按预期存在。主题主题已使用 TopicNameStrategy 设置，并且在 yml 配置中也提供了相同的内容。如果有人有任何想法，请告诉我

编辑：在此处添加配置

nabkafka:
  kafka:
    allow.auto.create.topics: false
    schema-registry:
      cache-size: 2048
      auto.register.schemas: false
      key-subject-name-strategy: Topic
      value-subject-name-strategy: Topic
      subject-name-strategy: Topic
      ssl:
        protocol: SSL
        key-store-location: file:${infrastructure.services.ssl.keyStorePath}
        key-store-password: ${infrastructure.services.ssl.keyStorePassword}
        key-password: ${infrastructure.services.ssl.keyStorePassword}
        trust-store-location: file:${infrastructure.services.ssl.trustStorePath}
        trust-store-password: ${infrastructure.services.ssl.trustStorePassword}
        trust-store-type: JKS
        key-store-type: JKS
    producer:
      acks: all
      key-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
      properties:
        auto.register.schemas: false
    ssl:
      key-store-location: file:${infrastructure.services.ssl.keyStorePath}
      key-store-password: ${infrastructure.services.ssl.keyStorePassword}
      key-password: ${infrastructure.services.ssl.keyStorePassword}
      trust-store-location: file:${infrastructure.services.ssl.trustStorePath}
      trust-store-password: ${infrastructure.services.ssl.trustStorePassword}

请注意，我们在 spring kafka 上使用了一个包装器，它在组织中甚至在我们的非产品环境中都可以正常工作。

【问题讨论】：

能否也分享一下配置kafka的Spring属性？
@SukhmeetSethi 刚刚在上面添加
有同样的问题，但只有当我尝试发送 ForkJoinPool.common-worker 线程时才会重现。

标签： java apache-kafka spring-kafka confluent-schema-registry

【解决方案1】：

我遇到了同样的问题。正如@Vichukano 所提到的，这是与运行任务的执行程序（在本例中为 ForkJoinPool）相关的类加载器问题。你如何定义你正在使用的DefaultKafkaProducerFactory？从spring-kafka v2.8.0 开始，他们added a flag 在创建新生产者时配置序列化程序（消费者工厂的相同功能）。您应该事先配置（如果需要）您的序列化程序并将此标志设置为 false，例如：

  @Bean
  ProducerFactory producerFactory(SchemaRegistryClient registryClient, KafkaProperties kafkaProperties) {
    var serializer = new KafkaAvroSerializer(registryClient);
    var configs = kafkaProperties.buildProducerProperties();

    serializer.configure(configs, false);

    return new DefaultKafkaProducerFactory(configs, new StringSerializer(), serializer, false);
  }

希望对你有帮助！

【讨论】：