[jira] [Created] (KAFKA-7365) max.poll.records setting in Kafka Consumer is not working

["Kashyap Ivaturi (JIRA)" <ji...@apache.org>]

Kashyap Ivaturi created KAFKA-7365:
--------------------------------------

             Summary: max.poll.records setting in Kafka Consumer is not working
                 Key: KAFKA-7365
                 URL: https://issues.apache.org/jira/browse/KAFKA-7365
             Project: Kafka
          Issue Type: Bug
          Components: consumer
            Reporter: Kashyap Ivaturi


Hi,

I have a requirement where I consume messages one by one, each message has additional processing that I should do and then manually commit the offset.

Things work well most of the times until I get a big bunch of records which takes longer time to process and I encounter CommitFailed exception for the last set of records even though they were processed. While i'am able to reconnect back its picking some messages that I had already processed. I don't want this to happen as its creating duplicates in target systems that I integrate with while processing the message.

 

I decided that even though there are more messages in the queue , I would like to have a control on how many records I can process when polled.

I tried to replicate a scenario where I have started the consumer by setting 'max.poll.records' to '1' and then pushed 4 messages into the Topic the consumer is listening.

I expected that the consumer will only process 1 message because of my 'max.poll.records' setting but the consumer has processed all the 4 messages in single poll. Any idea why did it not consider 'max.poll.records' setting or is some other setting overriding this setting?. Appreciate your help or guidance in troubleshooting this issue.

Here is the log of my Consumer config when it starts:

 

2018-08-28 08:29:47.873  INFO 91121 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values: 

[auto.commit.interval.ms|https://auto.commit.interval.ms/] = 5000

auto.offset.reset = earliest

bootstrap.servers = [messaging-rtp3.cisco.com:9093]

check.crcs = true

[client.id|https://client.id/] = 

[connections.max.idle.ms|https://connections.max.idle.ms/] = 540000

enable.auto.commit = false

exclude.internal.topics = true

fetch.max.bytes = 52428800

[fetch.max.wait.ms|https://fetch.max.wait.ms/] = 500

fetch.min.bytes = 1

[group.id|https://group.id/] = empestor

[heartbeat.interval.ms|https://heartbeat.interval.ms/] = 3000

interceptor.classes = null

internal.leave.group.on.close = true

isolation.level = read_uncommitted

key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer

max.partition.fetch.bytes = 1048576

[max.poll.interval.ms|https://max.poll.interval.ms/] = 300000

max.poll.records = 1

[metadata.max.age.ms|https://metadata.max.age.ms/] = 300000

metric.reporters = []

metrics.num.samples = 2

metrics.recording.level = INFO

[metrics.sample.window.ms|https://metrics.sample.window.ms/] = 30000

partition.assignment.strategy = [org.apache.kafka.clients.consumer.RangeAssignor]

receive.buffer.bytes = 65536

[reconnect.backoff.max.ms|https://reconnect.backoff.max.ms/] = 1000

[reconnect.backoff.ms|https://reconnect.backoff.ms/] = 50

[request.timeout.ms|https://request.timeout.ms/] = 40000

[retry.backoff.ms|https://retry.backoff.ms/] = 100

sasl.jaas.config = null

sasl.kerberos.kinit.cmd = /usr/bin/kinit

sasl.kerberos.min.time.before.relogin = 60000

sasl.kerberos.service.name = null

sasl.kerberos.ticket.renew.jitter = 0.05

sasl.kerberos.ticket.renew.window.factor = 0.8

sasl.mechanism = GSSAPI

security.protocol = SSL

send.buffer.bytes = 131072

[session.timeout.ms|https://session.timeout.ms/] = 10000

ssl.cipher.suites = null

ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]

ssl.endpoint.identification.algorithm = null

ssl.key.password = [hidden]

ssl.keymanager.algorithm = SunX509

ssl.keystore.location = /kafka/certs/empestor/certificates/kafka.client.empestor.keystore.jks

ssl.keystore.password = [hidden]

ssl.keystore.type = JKS

ssl.protocol = TLS

ssl.provider = null

ssl.secure.random.implementation = null

ssl.trustmanager.algorithm = PKIX

ssl.truststore.location = /kafka/certs/empestor/certificates/kafka.client.truststore.jks

ssl.truststore.password = [hidden]

ssl.truststore.type = JKS

value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer

 

2018-08-28 08:29:48.079  INFO 91121 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 1.0.0



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)