SetPollTimeOut and HeartBeat

[dilipthelip]

Hi ,

I have a two clarifications on the below piece of code.

Clarification 1:

We have the polltimeout set as follows.

@Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(Integer.valueOf(env.getProperty(Constants.KAFKA_CONCURRENCY)));
        factory.getContainerProperties().setPollTimeout(3000);
        factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);
        factory.setRetryTemplate(new RetryTemplate() {{
            setRetryPolicy(new AlwaysRetryPolicy());
            setBackOffPolicy(new FixedBackOffPolicy() {{
                setBackOffPeriod(1000); // back off for 1 second between retries
            }});
        }});

Is this above setPollTimeout similar to poll method in the Kafka consumer class ?

  KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
      
  ConsumerRecords<String, String> records = consumer.poll(100);

What exactly happens when the consumer comes up ?

Does the consumer polls the topic and waits for 3 seconds and pull the records and return ?

or

Does it poll the topic every 3 seconds?

factory.getContainerProperties().setPollTimeout(3000);

Clarification 2:

We are leveraging consumer group strategy in our consumers.When will the heartbeat be sent to group coordinator ?

Is the Heartbeat sent during poll method ? Is the Heart beat sent while commiting the offset ?

Any help is much appreciated ?

Thanks !

[artembilan]

1. Yes, that option is exactly for the Consumer.poll():

ConsumerRecords<K, V> records = this.consumer.poll(this.containerProperties.getPollTimeout());

and the JavaDocs on the matter:

* @param timeout The time, in milliseconds, spent waiting in poll if data is not available in the buffer.
*            If 0, returns immediately with any records that are available currently in the buffer, else returns empty.

So, we are blocked for that time if there is no data and go ahead immediately if there is enough records according max.poll.records:

private static final String MAX_POLL_RECORDS_DOC = "The maximum number of records returned in a single call to poll().";

2. In Kafka 0.9 the Heartbeat is really sent during poll(). The newer versions do that in the separate process: https://cwiki.apache.org/confluence/display/KAFKA/KIP-62%3A+Allow+consumer+to+send+heartbeats+from+a+background+thread

[dilipthelip]

@artembilan - Thanks for the response.

We are manually managing the offset using the AcknowledgingMessageListener.
In this case will there be any heart beat sent during the below call.

acknowledgment.acknowledge();

[artembilan]

No, acknowledgment.acknowledge(); only does this:

ListenerConsumer.this.acks.put(this.record);

Places the record for commit to the queue.
The heardbeat is done only during main loop on the Consumer.poll().

[dilipthelip]

@artembilan - Thanks.

Another quick question,
Consecutive polling to kafka topic call happens only after the records that were pulled as part of the poll are successfully processed right ?

[garyrussell]

Not with the current design (which was mandated by the heartbeat being run on the consumer thread). Now that 0.10.1.x runs the heartbeats in the background we can go back to running the listener on the consumer thread.

The current process will fetch a second and third batch and if it can't hand the third batch to the listener (because it is still busy with the first), we pause() the consumer. This is so we can keep polling, but not fetching any more records.

In the next release, we'll go back to running the listener on the consumer thread so, then we won't call poll until the the current batch is processed.

[dilipthelip]

@garyrussell - Our kafka cluster is 0.9.
The second and third batch happens at what intervals?

I believe it does not depend on the below settings ?

 setPollTimeout(3000);

[artembilan]

If your listener is fast enough, there is no delays in between polls.
The main consumer thread just shifts messages to the queue and go back to the main loop: https://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java#L523

Just follow the code!

[garyrussell]

The ConumerRecords is handed off to the listener thread and the poll() is immediately performed - this is required so that the lack of a heartbeat doesn't cause the broker to rebalance the partitions when the consumer is a bit slow. We transfer the batch via a queue which has a depth of 1; if the queue is still full when we get the third batch (because the listener is still processing batch1), we pause() the consumer, but still keep calling poll() to avoid the rebalance.

[dilipthelip]

@garyrussell @artembilan Thanks for all your inputs.

[artembilan]

Glad to feel useful!

Consider to close the issue since we have arranged all your concerns.

[sivaalli]

Thanks for the information guys but I have a small question. Does this max.poll.interval.ms have any significance since @garyrussell mentioned that we after third batch we pause the consumer but still call poll() to not cause rebalance.

[garyrussell]

It has significance only with versions 1.3.x and above because now the the listener is called on the consumer thread. The old mechanism was required with older clients; it is recommended to upgrade to 1.3.x or higher (current version is 2.1.6) because the threading model is much, much simpler and newer clients can talk to older brokers if you can't upgrade the broker.

Please ask questions on Stack Overflow, not GitHub issues, and especially not on old closed issues.