Блог

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

Есть ли способ добиться этого в потоках Spring boot Kafka?

Я смог добиться этого с помощью BindingsEndpoint .

 private final BindingsEndpoint binding;

@Override
public void stop() {
    List<?> objects = binding.queryStates();
    if (!objects.isEmpty()) {
        log.info("Stopping Kafka topics ");
        List<Binding> bindings = getBindings(objects);
        bindings.forEach(Binding::stop);
        log.info("Stopped Kafka topics ");
    }
}

@Override
public void start() {
    List<?> objects = binding.queryStates();
    if (!objects.isEmpty()) {
        log.info("Starting Kafka topics ");
        List<Binding> bindings = getBindings(objects);
        bindings.forEach(Binding::start);
        log.info("Started Kafka topics ");
    }
}

protected List<Binding> getBindings(List<?> objects) {
    return objects.stream().filter(object -> object instanceof Binding)
            .map(object -> (Binding) object).collect(Collectors.toList());
}