Java установил время удерживания Кафки в исходном коде

Question

У меня следующая проблема. Мне нужно установить время хранения в Кафке для определенных выбранных тем. Я нашел решение, в котором я могу установить его с помощью следующей команды:

kafka-topics --zookeeper localhost:2181 --alter --topic topic-name --config retention.ms=-1

Я зарегистрировался в веб-интерфейсе Kafka и подтвердил, что оно изменилось.

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

Можно ли вообще установить время хранения в Java и, если возможно, как я могу его получить? сделано?

Заранее спасибо!

Answer 1

Это работает для меня:)

    private void setRetentionTime(String topicName, int retentionTime) {
        ConfigResource resource = new ConfigResource(Type.TOPIC, topicName);

        Collection<ConfigEntry> entries = new ArrayList<>();
        entries.add(new ConfigEntry(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionTime)));

        Config config = new Config(entries);
        Map<ConfigResource, Config> configs = new HashMap<>();
        configs.put(resource, config);

        AdminClient client = kafkaConfig.createAdminClient();
        client.alterConfigs(configs);
    }

Answer 2

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

Недавно с новыми версиями Kafka они представили это, и я думаю, что это можно сделать.

https://kafka.apache.org/10/javadoc/org/apache/kafka/clients/admin/KafkaAdminClient.html#alterConfigs-java.util.Map-org.apache.kafka.clients.admin.AlterConfigsOptions-

Теперь ответ звучит немного расплывчато, но вы должны быть более конкретны, если хотите получить полный ответ. Я могу расширить ответ, если у вас есть точная версия Kafka, библиотеки, которые вы используете для подключения к kafka и т. Д. ...

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