MessageListener { Consumer<byte[], byte[]> createConsumer() { synchronized (this) { while (!shutdown) { if (uri.resolveURI()) { return consumer(); } logger.warn("failed to resolve kafka uri, retry in 10 seconds, uri={}", uri); Threads.sleepRoughly(Duration.ofSeconds(10)); } return null; } } MessageListener(KafkaURI uri, String name, LogManager logManager); void subscribe(String topic, Class<T> messageClass, MessageHandler<T> handler, BulkMessageHandler<T> bulkHandler); void start(); void shutdown(); void awaitTermination(long timeoutInMs); final ConsumerMetrics consumerMetrics; public int poolSize; public Duration maxProcessTime; public Duration longConsumerDelayThreshold; public int maxPollRecords; public int maxPollBytes; public int minPollBytes; public Duration maxWaitTime; public String groupId; }
@Test void createConsumer() { assertThat(listener.createConsumer()).isNotNull(); }