### Question:
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 { 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(); } }  @Override void configure(final Conf configs); @Override void onStart(final StreamsLifecycleContext context, final StreamsLifecycleChain chain);  }### Answer:
@Test public void shouldThrowIllegalStateExceptionWhenInterceptorIsNotConfigured() { Assertions.assertThrows(IllegalStateException.class, () -> new KafkaBrokerReadyInterceptor().onStart(null, null) ); }