Spring boot producer在kafka重新启动后无法发送任何消息

Question

spring-boot消费者微服务无法在kafka重新启动后向主题发送消息。

>

消费者和生产者（spring boot微服务）位于同一覆盖网络“Net-Broker”上，因此它们使用服务名“kafka:9092”访问kafka。

一开始一切都很顺利。

然后kafka仅被重新启动，在此之后，消费者不能再从kafka主题发送消息。

由于docker-compose.yml中的一个小更改（例如max_actempts:3=>max_actempts:4),kafka服务将重新启动

docker-compose.yml文件

kafka:
    image: wurstmeister/kafka:2.12-2.2.0
    depends_on:
      - zookeeper
    networks:
      - net-broker
    deploy:
      replicas: 1
      update_config:
        parallelism: 1
        delay: 10s
      restart_policy:
        condition: on-failure
        max_attempts: 3
    # ports:
    #   - target: 9094
    #     published: 9094
    #     protocol: tcp
    #     mode: host
    environment:
      HOSTNAME_COMMAND: "echo ${HOST_IP:-192.168.99.100}"
      KAFKA_CREATE_TOPICS: "gnss-topic-${GNSS_TAG}:3:1"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: INSIDE://:9092,OUTSIDE://_{HOSTNAME_COMMAND}:9094
      KAFKA_LISTENERS: INSIDE://:9092,OUTSIDE://:9094
      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
      BROKER_ID_COMMAND: "echo 101"
      KAFKA_LOG_DIRS: "/kafka/kafka-logs"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - kafka-data:/kafka

KafkaProducerConfig类

@Bean
  public ProducerFactory<String, GNSSPortHolderDTO> producerFactory() {
    Map<String, Object> configProps = new HashMap<>();

    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, gnssConfig.getKafkaBootstapServers());

    // high throughput producer (at the expense of a bit of latency and CPU usage)
    configProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
    configProps.put(ProducerConfig.LINGER_MS_CONFIG, "20");
    configProps.put(ProducerConfig.BATCH_SIZE_CONFIG, Integer.toString(32*1024)); // 32 KB batch size

    // serializers
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

    return new DefaultKafkaProducerFactory<>(configProps);
  }

org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for gnss-topic-11.2.1-B5607-1: 30030 ms has passed since batch creation plus linger time

gnss_data-access.1.ll948jogpqil@GDN-S-GNSS2    | 2019-05-08 09:42:33.984  INFO 1 --- [ gnss-view-data] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=gnss-view-data] Marking the coordinator fe7091944126:9092 (id: 2147483546 rack: null) dead

Answer 1

我找到了解决这个问题的办法。

只是提醒一下：

• 只能从Docker Net-Broker的覆盖虚拟网络访问Kafka。
• 出于安全原因，不应从Docker主机IP访问Kafka

旧值（重新启动后不起作用）:kafka_advertised_listeners:inside://:9092,outside:/_{HOSTNAME_COMMAND}:9094

新值（重新启动后工作）:kafka_advertised_listeners:inside://kafka:9092,outside:/_{HOSTNAME_COMMAND}:9094

因此，修复方法是为内部广告侦听器指定kafka服务名kafka:9092。

问题是，即使spring boot producer被配置为使用kafka:9092

configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka:9092);  

生产者实际上使用kafka容器ID进行通信，而不是使用服务名kafka:9092，因此在kafka重新启动后，创建了一个新容器（新容器ID），但生产者仍然指向旧容器ID