Hello I m using Flink DataStream 1 16 API to read from inbou Apache Flink #troubleshooting

Hello, I’m using Flink DataStream(1.16) API to rea...

sharad mishra

03/01/2023, 12:34 PM

Hello, I’m using Flink DataStream(1.16) API to read from inbound Kafka topic and writing to another outbound Kafka Topic. I’ve enabled

exactly once

deliveree guarantee in my program. I got following exception in my application, along with duplicate data in the Kafka sink. I’m curious to know what might have caused this exception along with duplicate data although deliveree guarantee is set to

exactlyonce

Copy code

org.apache.flink.util.FlinkRuntimeException: Failed to send data to Kafka dcn_audit_log_events_unique_stg-6@-1 with FlinkKafkaInternalProducer{transactionalId='unique-6-1222', inTransaction=true, closed=false} 
	at org.apache.flink.connector.kafka.sink.KafkaWriter$WriterCallback.throwException(KafkaWriter.java:436)
	at org.apache.flink.connector.kafka.sink.KafkaWriter$WriterCallback.lambda$onCompletion$0(KafkaWriter.java:417)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
	at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:398)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:367)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:352)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:831)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:780)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:914)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550)
	at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.kafka.common.errors.InvalidProducerEpochException: Producer attempted to produce with an old epoch.

Here is the snapshot of FlinkKafkaConsumer properties from taskmanager logs:

Copy code

acks = -1
	batch.size = 16384
	bootstrap.servers = [10.7.111.32:9092]
	buffer.memory = 33554432
	client.dns.lookup = use_all_dns_ips
	client.id = producer-duplicate-13-402
	compression.type = none
	<http://connections.max.idle.ms|connections.max.idle.ms> = 540000
	<http://delivery.timeout.ms|delivery.timeout.ms> = 120000
	enable.idempotence = true
	interceptor.classes = []
	key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
	<http://linger.ms|linger.ms> = 0
	<http://max.block.ms|max.block.ms> = 60000
	max.in.flight.requests.per.connection = 5
	max.request.size = 1048576
	<http://metadata.max.age.ms|metadata.max.age.ms> = 300000
	<http://metadata.max.idle.ms|metadata.max.idle.ms> = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	<http://metrics.sample.window.ms|metrics.sample.window.ms> = 30000
	partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
	receive.buffer.bytes = 32768
	<http://reconnect.backoff.max.ms|reconnect.backoff.max.ms> = 1000
	<http://reconnect.backoff.ms|reconnect.backoff.ms> = 50
	<http://request.timeout.ms|request.timeout.ms> = 30000
	retries = 2147483647
	<http://retry.backoff.ms|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
	<http://sasl.login.connect.timeout.ms|sasl.login.connect.timeout.ms> = null
	<http://sasl.login.read.timeout.ms|sasl.login.read.timeout.ms> = 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
	<http://sasl.login.retry.backoff.max.ms|sasl.login.retry.backoff.max.ms> = 10000
	<http://sasl.login.retry.backoff.ms|sasl.login.retry.backoff.ms> = 100
	sasl.mechanism = GSSAPI
	sasl.oauthbearer.clock.skew.seconds = 30
	sasl.oauthbearer.expected.audience = null
	sasl.oauthbearer.expected.issuer = null
	<http://sasl.oauthbearer.jwks.endpoint.refresh.ms|sasl.oauthbearer.jwks.endpoint.refresh.ms> = 3600000
	<http://sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms|sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms> = 10000
	<http://sasl.oauthbearer.jwks.endpoint.retry.backoff.ms|sasl.oauthbearer.jwks.endpoint.retry.backoff.ms> = 100
	sasl.oauthbearer.jwks.endpoint.url = null
	sasl.oauthbearer.scope.claim.name = scope
	sasl.oauthbearer.sub.claim.name = sub
	sasl.oauthbearer.token.endpoint.url = null
	security.protocol = PLAINTEXT
	security.providers = null
	send.buffer.bytes = 131072
	<http://socket.connection.setup.timeout.max.ms|socket.connection.setup.timeout.max.ms> = 30000
	<http://socket.connection.setup.timeout.ms|socket.connection.setup.timeout.ms> = 10000
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2]
	ssl.endpoint.identification.algorithm = https
	ssl.engine.factory.class = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.certificate.chain = null
	ssl.keystore.key = null
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLSv1.2
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.certificates = null
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	<http://transaction.timeout.ms|transaction.timeout.ms> = 600000
	transactional.id = duplicate-13-402
	value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer

also FlinkKafkaProducer properties snapshot:

Copy code

org.apache.kafka.clients.producer.ProducerConfig - ProducerConfig values: 
	acks = -1
	batch.size = 16384
	bootstrap.servers = [10.7.111.32:9092]
	buffer.memory = 33554432
	client.dns.lookup = use_all_dns_ips
	client.id = producer-duplicate-4-402
	compression.type = none
	<http://connections.max.idle.ms|connections.max.idle.ms> = 540000
	<http://delivery.timeout.ms|delivery.timeout.ms> = 120000
	enable.idempotence = true
	interceptor.classes = []
	key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
	<http://linger.ms|linger.ms> = 0
	<http://max.block.ms|max.block.ms> = 60000
	max.in.flight.requests.per.connection = 5
	max.request.size = 1048576
	<http://metadata.max.age.ms|metadata.max.age.ms> = 300000
	<http://metadata.max.idle.ms|metadata.max.idle.ms> = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	<http://metrics.sample.window.ms|metrics.sample.window.ms> = 30000
	partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
	receive.buffer.bytes = 32768
	<http://reconnect.backoff.max.ms|reconnect.backoff.max.ms> = 1000
	<http://reconnect.backoff.ms|reconnect.backoff.ms> = 50
	<http://request.timeout.ms|request.timeout.ms> = 30000
	retries = 2147483647
	<http://retry.backoff.ms|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
	<http://sasl.login.connect.timeout.ms|sasl.login.connect.timeout.ms> = null
	<http://sasl.login.read.timeout.ms|sasl.login.read.timeout.ms> = 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
	<http://sasl.login.retry.backoff.max.ms|sasl.login.retry.backoff.max.ms> = 10000
	<http://sasl.login.retry.backoff.ms|sasl.login.retry.backoff.ms> = 100
	sasl.mechanism = GSSAPI
	sasl.oauthbearer.clock.skew.seconds = 30
	sasl.oauthbearer.expected.audience = null
	sasl.oauthbearer.expected.issuer = null
	<http://sasl.oauthbearer.jwks.endpoint.refresh.ms|sasl.oauthbearer.jwks.endpoint.refresh.ms> = 3600000
	<http://sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms|sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms> = 10000
	<http://sasl.oauthbearer.jwks.endpoint.retry.backoff.ms|sasl.oauthbearer.jwks.endpoint.retry.backoff.ms> = 100
	sasl.oauthbearer.jwks.endpoint.url = null
	sasl.oauthbearer.scope.claim.name = scope
	sasl.oauthbearer.sub.claim.name = sub
	sasl.oauthbearer.token.endpoint.url = null
	security.protocol = PLAINTEXT
	security.providers = null
	send.buffer.bytes = 131072
	<http://socket.connection.setup.timeout.max.ms|socket.connection.setup.timeout.max.ms> = 30000
	<http://socket.connection.setup.timeout.ms|socket.connection.setup.timeout.ms> = 10000
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2]
	ssl.endpoint.identification.algorithm = https
	ssl.engine.factory.class = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.certificate.chain = null
	ssl.keystore.key = null
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLSv1.2
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.certificates = null
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	<http://transaction.timeout.ms|transaction.timeout.ms> = 600000
	transactional.id = duplicate-4-402
	value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer

5 Views

Open in Slack

Previous Next