Kafka listener isn't listening to Kafka

Viewed 405

I'm using Java 11 and kafka-client 2.0.0.

I'm using the following code to generate a consumer :

    public Consumer createConsumer(Properties properties,String regex) {
        log.info("Creating consumer and listener..");
        Consumer consumer = new KafkaConsumer<>(properties);
        ConsumerRebalanceListener listener = new ConsumerRebalanceListener() {

            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                log.info("The following partitions were revoked from consumer : {}", Arrays.toString(partitions.toArray()));
            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                log.info("The following partitions were assigned to consumer : {}", Arrays.toString(partitions.toArray()));
            }
        };
        consumer.subscribe(Pattern.compile(regex), listener);
        log.info("consumer subscribed");
        return consumer;
    }
}

My poll loop is in a different place in the code :

public <K, V> void startWorking(Consumer<K, V> consumer) {
        try {
            while (true) {
                ConsumerRecords<K, V> records = consumer.poll(600);
                if (records.count() > 0) {
                    log.info("Polled {} records", records.count());

                } else {
                    log.info("polled 0 records.. going to sleep..");
                    Thread.sleep(200);
                }
            }
        } catch (WakeupException | InterruptedException e) {
            log.error("Consumer is shutting down", e);
        } finally {
            consumer.close();
        }
    }

When I run the code and use this function, the consumer is created and the log contains the following messages :

    Creating consumer and listener..
    consumer subscribed
polled 0 records.. going to sleep..
polled 0 records.. going to sleep..
polled 0 records.. going to sleep..

The log doesn't contain any info regarding the partition assignment/revocation.

In addition I'm able to see in the log the properties that the consumer uses (group.id is set) :

2020-07-09 14:31:07.959 DEBUG 7342 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values:
        auto.commit.interval.ms = 5000
        auto.offset.reset = latest
        bootstrap.servers = [server1:9092]
        check.crcs = true
        client.id =
        group.id=mygroupid
        key.deserializer=..
        value.deserializer=..

So I tried to use the kafka-console-consumer with the same configuration in order to consume from one of the topics that the regex(mytopic.*) should catch (in this case I used the topic mytopic-1) :

/usr/bin/kafka-console-consumer.sh --bootstrap-server server1:9092 --topic mytopic-1 --property print.timestamp=true --consumer.config /data/scripts/kafka-consumer.properties  --from-begining

I have a poll loop in other part of my code that is timing out every 10m .So the bottom line - the problem is that partitions aren't assigned to the Java consumer. The prints inside the listener never happen and the consumer doesn't have any partitions to listen to.

1 Answers

It seems that I was missing the ssl property in my properties file. Don't forget to specify the security.protocol=ssl if you use ssl. It seems that kafka-client API doesn't throw exception if the Kafka uses ssl and you try to access it without ssl parameter configured.

Related