Java Kafka adminClient настройка темы.Значения конфигурации перезаписываются

Question

При попытке настроить вновь созданную тему kafka, используя java kafka adminClient, значения перезаписываются.

Я попытался настроить ту же конфигурацию темы с помощью консольных команд, и она работает.К сожалению, когда я пытаюсь выполнить код Java, некоторые значения сталкиваются и перезаписываются.

ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
Map<ConfigResource, Config> updateConfig = new HashMap<>();

// update retention Bytes for this topic
ConfigEntry retentionBytesEntry = new ConfigEntry(TopicConfig.RETENTION_BYTES_CONFIG, String.valueOf(retentionBytes));
updateConfig.put(resource, new Config(Collections.singleton(retentionBytesEntry)));

// update retention ms for this topic
ConfigEntry retentionMsEntry = new ConfigEntry(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionMs));
updateConfig.put(resource, new Config(Collections.singleton(retentionMsEntry)));

// update segment Bytes for this topic
ConfigEntry segmentBytesEntry = new ConfigEntry(TopicConfig.SEGMENT_BYTES_CONFIG, String.valueOf(segmentbytes));
updateConfig.put(resource, new Config(Collections.singleton(segmentBytesEntry)));

// update segment ms for this topic
ConfigEntry segmentMsEntry = new ConfigEntry(TopicConfig.SEGMENT_MS_CONFIG, String.valueOf(segmentMs));
updateConfig.put(resource, new Config(Collections.singleton(segmentMsEntry)));

// Update the configuration
client.alterConfigs(updateConfig);

Я ожидаю, что в теме правильно заданы все заданные значения конфигурации.

Answer 1

Ваша логика не работает правильно, потому что вы звоните Map.put() несколько раз с одним и тем же ключом.Следовательно, сохраняется только последняя запись.

Правильный способ указать несколько конфигураций тем - добавить их в объект ConfigEntry.Только после добавления ConfigEntry к Map.

Например:

// Your Topic Resource
ConfigResource cr = new ConfigResource(Type.TOPIC, "mytopic");

// Create all your configurations
Collection<ConfigEntry> entries = new ArrayList<>();
entries.add(new ConfigEntry(TopicConfig.SEGMENT_BYTES_CONFIG, String.valueOf(segmentbytes)));
entries.add(new ConfigEntry(TopicConfig.RETENTION_BYTES_CONFIG, String.valueOf(retentionBytes)));
...

// Create the Map
Config config = new Config(entries);
Map<ConfigResource, Config> configs = new HashMap<>();
configs.put(cr, config);

// Call alterConfigs()
admin.alterConfigs(configs);