Hello,
Since the last update to the universal Kafka connector, I'm getting the
following error fairly often.
/2019-11-18 15:42:52,689 ERROR
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer
clientId=consumer-10, groupId=srx-consumer-group] Offset commit failed on
partition events-4 at offset 38173628004: The request timed out.
&unknown 2019-11-18 15:42:52,707 WARN
org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher -
Committing offsets to Kafka failed. This does not compromise Flink's
checkpoints.
&unknown 2019-11-18 15:42:52,707 WARN
org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase - Async
Kafka commit failed.
red 2019-11-18 15:42:52,689 ERROR
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer
clientId=consumer-10, groupId=srx-consumer-group] Offset commit failed on
partition events-4 at offset 38173628004: The request timed out.
yellow 2019-11-18 15:42:52,707 WARN
org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher -
Committing offsets to Kafka failed. This does not compromise Flink's
checkpoints.
yellow 2019-11-18 15:42:52,707 WARN
org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase - Async
Kafka commit failed./
---
I've been investigating and I can't find what is causing this. We've been
monitoring several metrics like the recordsConsumed, fetch-size-avg and
fetch-rate but the values are the same when the error happens and when it
doesn't. So we know there isn't a peak of events or a larger fetched size
when the problem occurs. We also monitor other metrics like CPU, Memory,
GCs, Network IO, Network connections and Disk IO but we haven't found
anything out of the ordinary.
Our job has two source nodes reading from two distinct kafka topics, the
problem happens on both source nodes.
*Flink Version: *1.8.2
*Kafka Version:* 2.3.0
*My kafka consumer Properties:*
/2019-11-14 16:51:20,142 INFO
org.apache.kafka.clients.consumer.ConsumerConfig -
ConsumerConfig values:
auto.commit.interval.ms = 5000
auto.offset.reset = latest
bootstrap.servers = [hidden]
check.crcs = true
client.id =
connections.max.idle.ms = 540000
default.api.timeout.ms = 60000
enable.auto.commit = false
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
fetch.min.bytes = 1
group.id = srx-consumer-group
heartbeat.interval.ms = 3000
interceptor.classes = []
internal.leave.group.on.close = true
isolation.level = read_uncommitted
key.deserializer = class
org.apache.kafka.common.serialization.ByteArrayDeserializer
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 300000
max.poll.records = 500
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partition.assignment.strategy = [class
org.apache.kafka.clients.consumer.RangeAssignor]
receive.buffer.bytes = 65536
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 30000
retry.backoff.ms = 100
sasl.client.callback.handler.class = null
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.login.callback.handler.class = null
sasl.login.class = null
sasl.login.refresh.buffer.seconds = 300
sasl.login.refresh.min.period.seconds = 60
sasl.login.refresh.window.factor = 0.8
sasl.login.refresh.window.jitter = 0.05
sasl.mechanism = GSSAPI
security.protocol = SSL
send.buffer.bytes = 131072
session.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = https
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = /etc/pki/java/ca.ks
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 = /etc/pki/java/ca.ks
ssl.truststore.password = [hidden]
ssl.truststore.type = JKS
value.deserializer = class
org.apache.kafka.common.serialization.ByteArrayDeserializer
/
*
Commit latency vs commits failed:*
<http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/file/t612/Screenshot_2019-11-26_at_09.png>
We have checkpointing enabled (synchronous checkpoints every 10 secs). We
also have the job configured to commit kafka offsets on each checkpoint
(default behaviour).
-----
Best Regards,
Pedro Chaves
--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
