log.retention和cleanup.policy - Spring Kafka

Question

我的docker-compose.yml中有以下配置

kafka:
    image: 'confluentinc/cp-kafka:6.1.0'
    hostname: kafka
    container_name: kafka
    ports:
      - '9092:9092'
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://127.0.0.1:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_LOG_RETENTION_MINUTES: 5
      KAFKA_LOG_CLEANUP_POLICY: compact, delete
    depends_on:
      - zookeeper

我使用Spring Kafka Binder Streams来创建主题。一旦应用程序启动，我可以在kafka中看到主题，但我看不到配置的log.retention.minutes或cleanup.policy。

如果我指定了特定主题的配置，它就会起作用。不能在代理级别配置日志保留和清理吗？

    @Bean
    public NewTopic crateEmployeeTopic() {
        return TopicBuilder
                .name(KafkaConfig.EMP_DATA_TOPIC)
                .partitions(3)
                .config(TopicConfig.RETENTION_MS_CONFIG, "90000")
                .config(TopicConfig.CLEANUP_POLICY_CONFIG, "compact, delete")
                .build();
    }

-更新--

我使用Kafdrop UI来检查主题配置。当它在主题级别定义时，它显示了配置，但在代理级别，它没有显示任何配置。我运行describe命令进行了仔细检查，现在可以看到cleanup.policy，但retention仍然缺失

$ kafka-topics --bootstrap-server localhost:9092 --describe -- topic  emp_data
Topic: emp_data    PartitionCount: 3      ReplicationFactor: 1    Configs: cleanup.policy=compact,delete
        Topic: emp_data   Partition: 0    Leader: 1       Replicas: 1     Isr: 1

谢谢您抽时间见我。

Answer 1

这个命令起到了作用。

$ kafka-configs --bootstrap-server localhost:9092 --entity-type brokers --entity-name 1 --all --describe | grep retention

Answer 2

您可以像在here中提到的那样，为log.retention.minutes设置代理配置。然后，所有配置的主题都默认使用它。

cleanup.policy也一样。Read here获取更多信息。