我需要在创建过程中配置特定主题的保留策略。我试图寻找解决方案,我只能找到如下所示的命令级别alter命令
./bin/kafka-topics.sh –zookeeper本地主机:2181-更改–topic我的主题–config保留.ms = 1680000
有人可以让我知道一种在创建过程中进行配置的方法,例如spring-mvc中的xml或属性文件配置。
Spring Kafka允许您通过@Bean在应用程序上下文中声明s来创建新主题。这将需要KafkaAdmin在应用程序上下文中具有类型的Bean,如果使用Spring Boot,则会自动创建该类型的Bean 。您可以如下定义主题:
@Bean
KafkaAdmin
@Bean public NewTopic myTopic() { return TopicBuilder.name("my-topic") .partitions(4) .replicas(3) .config(TopicConfig.RETENTION_MS_CONFIG, "1680000") .build(); }
如果您不使用Spring Boot,则还必须定义KafkaAdminbean:
@Bean public KafkaAdmin admin() { Map<String, Object> configs = new HashMap<>(); configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092"); return new KafkaAdmin(configs); }
如果您要编辑现有主题的配置,则必须使用AdminClient,这是retention.ms在主题级别更改的代码段:
AdminClient
retention.ms
Map<String, Object> config = new HashMap<>(); config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092"); AdminClient client = AdminClient.create(config); ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, "new-topic"); // Update the retention.ms value ConfigEntry retentionEntry = new ConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "1680000"); Map<ConfigResource, Config> updateConfig = new HashMap<ConfigResource, Config>(); updateConfig.put(resource, new Config(Collections.singleton(retentionEntry))); AlterConfigsResult alterConfigsResult = client.alterConfigs(updateConfig); alterConfigsResult.all();