Удаление логов kafka с использованием времени хранения - PullRequest
Новая
       33

Удаление логов kafka с использованием времени хранения

0 голосов
/ 15 февраля 2019

Я работаю на платформе Windows, инструменте, в котором я очень часто буду генерировать сообщения для kafka из приложения весенней загрузки (функция будет генерировать сообщения непрерывно).И я буду потреблять сообщения из приложения node.js.Таким образом, приложение создает так много тем в день, журналы Кафки занимают все дисковое пространство в течение недели.Поэтому я попытался с помощью функции log.retention.hours удалить журналы, но я получаю Ошибка при удалении сегментов , java.nio.file.FileSystemException: процесс не может получить доступфайл, потому что он используется другим процессом .

ПРИМЕЧАНИЕ. У меня нет решения, чтобы это исправить, я не знаю причину, по которой это происходит.У меня есть два вопроса

1) Нужно ли мне настроить какую-либо вещь в моем приложении или мне нужно отправить подтверждение на сервер kafka из моего весеннего загрузочного приложения, что я закончил создавать сообщения в тему, чтобы kafka удалилit.

2) как я могу подключиться к серверу kafka с других компьютеров? (я размещаю kafka, zookeeper на машине, и я создаю сообщения на тему с той же машины. Теперь я пытаюсь потреблятьсообщения с другой машины, но я не смог подключиться к серверу kafka)

Ниже приведены настройки, которые я использую в приложении весенней загрузки, и я буду создавать сообщения по темам. 

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}
...