Graceful shutdown of kafka (with other services)

[bojanv55]

In my code I use kafka for processing and then saving in the redis.

@Incoming("words-in")
    public Uni<Void> sink(String word) {
        return Uni.createFrom().item(word)
                .onItem().delayIt().by(Duration.ofSeconds(1))
                .onCancellation()
                .invoke(() -> System.out.println("⬆️ Cancelled"))
                .onTermination()
                .invoke(() -> System.out.println("🔚 Uni terminated"))
                .flatMap(w -> kv.set("lastv", w)) //<------- THIS IS REDIS COMMAND
                .onItem()
                .invoke(() -> {
                    System.out.println("Received word: " + word);});
    }

Whenever there is shutdown, I get this kind of exception:

2025-08-21 11:04:47,790 WARN  [io.sma.rea.mes.kafka] (vert.x-eventloop-thread-10) SRMSG18228: A failure has been reported for Kafka topics '[words]': java.lang.IllegalStateException: Client is closed
	at io.vertx.core.net.impl.NetClientImpl.connect(NetClientImpl.java:215)
	at io.vertx.core.net.impl.NetClientImpl.connect(NetClientImpl.java:127)
	at io.vertx.redis.client.impl.RedisConnectionManager$RedisConnectionProvider.connect(RedisConnectionManager.java:167)
	at io.vertx.core.net.impl.pool.SimpleConnectionPool.connect(SimpleConnectionPool.java:253)
	at io.vertx.core.net.impl.pool.SimpleConnectionPool$Acquire$3.run(SimpleConnectionPool.java:595)
	at io.vertx.core.net.impl.pool.Task.runNextTasks(Task.java:43)
	at io.vertx.core.net.impl.pool.CombinerExecutor.submit(CombinerExecutor.java:91)

It happens because redis has closed the connection, while there are still messages in-flight from kafka.

Is there some way to prevent redis from closing the connections before the last kafka message that has been ACK-ed is processed?

[cescoffier]

Kafka pulls many messages at once.
What you need to do is detect the shutdown (using an @observe ShutdownEvent with the right priority) and set a global flag. When the shutdown is detected, in your pipeline, skip the Redis command (probably throws an error)

[bojanv55]

I added something like this:

@GracefulShutdown
@Interceptor
public class GracefulShutdownInterceptor {

  @Inject GracefulShutdownCounter counter;

  @AroundInvoke
  public Object around(InvocationContext ctx) throws Exception {
    counter.increment();
    try {
      Object result = ctx.proceed();

      if (result instanceof Uni) {
        return ((Uni<?>) result).onTermination().invoke(() -> counter.decrement());
      } else if (result instanceof Multi) {
        return ((Multi<?>) result).onTermination().invoke(() -> counter.decrement());
      } else {
        counter.decrement();
        return result;
      }
    } catch (Exception e) {
      counter.decrement();
      throw e;
    }
  }
}

and

@ApplicationScoped
public class GracefulShutdownObserver {
  public static final long GRACEFUL_TWO_SECONDS_NANOS = 2_000_000_000L;
  public static final Logger LOGGER = LoggerFactory.getLogger(GracefulShutdownObserver.class);

  @Inject GracefulShutdownCounter counter;

  @Inject
  @Connector("smallrye-kafka")
  KafkaConnector connector;

  void onShutdown(@Observes @Priority(GRACEFUL_SHUTDOWN) ShutdownEvent event) {
    int logCounter = 0;

    connector.terminate(new Object());

    while (counter.hasInFlightMessages()) {
      try {
        if (logCounter % 10 == 0) {
          int inFlight = counter.get();
          if (inFlight > 0) {
            LOGGER.info(
                "There are still %s message(s) in-flight, waiting...".formatted(counter.get()));
          } else {
            LOGGER.info("In-flight queue empty, waiting for graceful shutdown period...");
          }
        }
        logCounter++;
        Thread.sleep(100);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
    }

    LOGGER.info("No in-flight messages, shutting down now...");
  }
}

Together with graceful-shutdown setting in properties, it allows to consume all of the messages before shutdown. If I put graceful-shutdown to false, it allows last message to be consumed and then it stops the process. I also tested that in this case, redis successfully updates the data, but last message delivered from kafka is not ACK-ed and it gets re-delivered on the next run. Is it possible somehow to ACK also that last message, since it gets processed?

[bojanv55]

fyi @cescoffier I had to call on KafkaConnector method connector.terminate(new Object()); because no matter what priority I gave to my observer, kafka would only invoke at the last step - it would just keep consuming the messages and not terminating the process. This way I notify it to do termination immediately (either graceful or not). Do you thing this is ok to do at this point?