class KafkaBrokerReadyInterceptor implements StreamsLifecycleInterceptor, Configurable {

    @Override
    public void onStart(final StreamsLifecycleContext context, final StreamsLifecycleChain chain) {
        if (config == null) throw new IllegalStateException("interceptor not configured");

        final State previousState = context.streamsState();

        boolean isReady = false;
        int numBrokerAvailable = 0;
        try {
            LOG.info("Checking for Kafka to be ready. Expected broker(s): {}", config.getMinAvailableBrokers());
            final var container = context.container();
            final var adminClient = container.getAdminClient();

            final var start = Time.SYSTEM.milliseconds();
            var remaining = config.getTimeoutMs();

            while (remaining > 0) {
                context.setState(InterceptorState.WAITING_FOR_KAFKA_BROKERS_READY);
                try {
                    Collection<Node> nodes = getClusterNodes(adminClient, remaining);
                    numBrokerAvailable = nodes.size();
                    if (isReady = !nodes.isEmpty() && numBrokerAvailable >= config.getMinAvailableBrokers()) {
                        break;
                    }
                } catch (Exception e) {
                    LOG.error("Unexpected error while listing Kafka nodes", e);
                }
                Time.SYSTEM.sleep(Duration.ofMillis(Math.min(config.getRetryBackoffMs(), remaining(start))));

                LOG.info("Waiting for Kafka cluster to be ready. Expected {} brokers, but only {} was found.",
                    config.getMinAvailableBrokers(),
                    numBrokerAvailable
                );
                remaining = remaining(start);
            }

        } finally {
            // always reset the previous state.
            context.setState(previousState);
            if (!isReady) {
                LOG.warn(
                    "Kafka Cluster is not ready yet. Expected {} brokers, but only {} was found. TimeoutMs expires",
                    config.getMinAvailableBrokers(),
                    numBrokerAvailable
                );
            }
            chain.execute();
        }
    }

}