如何配置 Spring Cloud Sleuth kafka zipkin sender 以减少保留期？答案

【问题标题】：How do you configure Spring Cloud Sleuth kafka zipkin sender to reduce the retention period?如何配置 Spring Cloud Sleuth kafka zipkin sender 以减少保留期？
【发布时间】：2022-02-22 03:45:02
【问题描述】：

我正在尝试进行设置，以便在部署新环境时，spring cloud sleuth zipkin 发件人使用的zipkinKafka 主题的保留期较短，因为它会被其他东西处理，否则会被忽略（在开发者机器上）。

我尝试添加这个

@Bean("zipkin")
public NewTopic zipkin() {

  return TopicBuilder.name("zipkin")
    .config(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(Duration.ofMinutes(2).getSeconds() * 1000))
    .compact()
    .replicas(1)
    .build();
}

但这不起作用，因为主题已经创建并且已经配置。我知道我能做到

kafka-configs --bootstrap-server $_KAFKA_BOOTSTRAP_SERVER --alter --topic zipkin --add-config retention.ms=2000

但我希望在 Spring 应用程序上设置它。

【问题讨论】：

标签： java spring apache-kafka spring-kafka

【解决方案1】：

您需要使用 AdminClient 实例来修改现有主题配置

【讨论】：

【解决方案2】：

Spring-Kafka 没有在应用程序上下文中提供AdminClient，因此需要构建它而不是让它为您自动装配。以下代码展示了如何通过使用环境中提供的引导服务器创建AdminClient 并设置多个属性来实现。

    /**
     * Retention time for zipkin messages.  Defaults to 2 minutes.
     */
    @Value("${zipkin.retention.ms:120000}")
    private long zipkinRetentionMs;

    /**
     * Segment bytes size for zipkin.  Defaults to 10MB.
     */
    @Value("${zipkin.segment.bytes:10485760}")
    private long zipkinSegmentBytes;

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @PostConstruct
    public void fixZipkin() {

        try (final var adminClient = AdminClient.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers))) {
            adminClient.incrementalAlterConfigs(
                Map.of(
                    new ConfigResource(ConfigResource.Type.TOPIC, "zipkin"),
                    List.of(
                        new AlterConfigOp(
                            new ConfigEntry(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(zipkinRetentionMs)),
                            AlterConfigOp.OpType.SET
                        ),
                        new AlterConfigOp(
                            new ConfigEntry(TopicConfig.SEGMENT_BYTES_CONFIG, String.valueOf(zipkinSegmentBytes)),
                            AlterConfigOp.OpType.SET
                        )
                    )));
        } finally {

        }

    }

【讨论】：