[
https://issues.apache.org/jira/browse/KAFKA-10335?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17530152#comment-17530152
]
Sean Chang commented on KAFKA-10335:
------------------------------------
hi,how to resovle the problem?it is a perfmance issue with producer？
> Blocking of producer IO thread when calling send() from callback
> ----------------------------------------------------------------
>
> Key: KAFKA-10335
> URL: https://issues.apache.org/jira/browse/KAFKA-10335
> Project: Kafka
> Issue Type: Bug
> Components: clients, producer
> Reporter: Alexander Sibiryakov
> Priority: Minor
>
> We had application which supposed to be using KafkaProducer to deliver
> results of some work. Sometimes delivery of results weren't successful
> because of network connectivity errors or maintenance happening on the broker
> side. In such cases we wanted application to send an event with error and
> original message details. All good, but we wanted errors to be delivered to a
> separate topic. So we implemented a callback in send() method, using the same
> producer instance and calling send() from there.
> This application worked for some time, but then we encountered that its
> producer was stuck. Almost no CPU consumption and expiring batches for hours.
> After connecting with debugger it turned out that sender IO thread is
> blocking. When record is expired, a callback was called, which contained a
> call to send(), implying usage of a new topic, which metadata is not present
> in producer's client cache. When send() is missing metadata, it is allowed to
> block for up to max.block.ms interval, which is 60 secs by default. If
> application is active, then it will quickly result in a large amount of
> accumulated records. Every record will block IO thread for 60s. Therefore
> sender IO thread is essentially blocked. In Producer only Sender IO thread
> contains a call to client's poll() method, which is responsible for all the
> network communication and metadata update. If poll() is executed with
> significant delay then it will result to errors, connected with various
> timeouts. That's it we've got a stuck producer with little chance to recover.
> To summarise, pre-requisites for the problem are sending from callback, using
> the same producer instance and usage of topic which wasn't seen before.
> I think it is important to decide if the issue is KafkaProducer misuse or its
> bug. Code is callbacks shouldn't block, that is clear, but at the same time,
> no one expects already initialised producer to block.
> Depending on decision I could produce a fix, it can be either a warning when
> user is trying to call a send() from callback, or reduction of max allowed
> blocking time for metadata update. It could be just docs changes, or even
> nothing.
> Here is code to reproduce the issue, the output it is producing follows the
> code snippet. Tested on Confluent Cloud, from my desktop with 100 Mbps
> connection.
> {code:java}
> public static void main(String[] args) throws IOException {
> byte[] blob = new byte[262144];
> Properties properties = new Properties();
> properties.load(new FileReader("kafka-staging.properties"));
> properties.setProperty("key.serializer",
> "org.apache.kafka.common.serialization.StringSerializer");
> properties.setProperty("value.serializer",
> "org.apache.kafka.common.serialization.ByteArraySerializer");
> properties.setProperty("request.timeout.ms", "5000");
> properties.setProperty("delivery.timeout.ms", "5000");
> KafkaProducer<String, byte[]> producer = new
> KafkaProducer(properties);
> while (true) {
> ProducerRecord<String, byte[]> record = new
> ProducerRecord<>("alex-test-valid-data", blob);
> producer.send(record, new Callback() {
> @Override
> public void onCompletion(RecordMetadata metadata, Exception
> exception) {
> if (exception != null) {
> System.err.println(exception);
> long start = System.currentTimeMillis();
> ProducerRecord<String, byte[]> record = new
> ProducerRecord<>("alex-test-errors", blob);
> producer.send(record); // blocking caused by
> metadata update
> long timeElapsed = System.currentTimeMillis() - start;
> System.err.println("time spent blocking IO thread: "
> + timeElapsed);
> }
> }
> });
> }
> }
> {code}
> {noformat}
> [2020-07-31 14:35:51,936: INFO/main] (AbstractConfig.java:347) -
> ProducerConfig values:
> acks = 1
> batch.size = 16384
> bootstrap.servers = [pkc-l915e.europe-west1.gcp.confluent.cloud:9092]
> buffer.memory = 33554432
> client.dns.lookup = default
> client.id =
> compression.type = none
> connections.max.idle.ms = 540000
> delivery.timeout.ms = 5000
> enable.idempotence = false
> interceptor.classes = []
> key.serializer = class
> org.apache.kafka.common.serialization.StringSerializer
> linger.ms = 0
> max.block.ms = 60000
> max.in.flight.requests.per.connection = 5
> max.request.size = 1048576
> metadata.max.age.ms = 300000
> metric.reporters = []
> metrics.num.samples = 2
> metrics.recording.level = INFO
> metrics.sample.window.ms = 30000
> partitioner.class = class
> org.apache.kafka.clients.producer.internals.DefaultPartitioner
> receive.buffer.bytes = 32768
> reconnect.backoff.max.ms = 1000
> reconnect.backoff.ms = 50
> request.timeout.ms = 5000
> retries = 2147483647
> retry.backoff.ms = 100
> sasl.client.callback.handler.class = null
> sasl.jaas.config = [hidden]
> 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 = PLAIN
> security.protocol = SASL_SSL
> security.providers = null
> send.buffer.bytes = 131072
> 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 = null
> ssl.keystore.password = null
> ssl.keystore.type = JKS
> ssl.protocol = TLS
> ssl.provider = null
> ssl.secure.random.implementation = null
> ssl.trustmanager.algorithm = PKIX
> ssl.truststore.location = null
> ssl.truststore.password = null
> ssl.truststore.type = JKS
> transaction.timeout.ms = 60000
> transactional.id = null
> value.serializer = class
> org.apache.kafka.common.serialization.ByteArraySerializer
> [2020-07-31 14:35:52,099: INFO/main] (AbstractLogin.java:61) - Successfully
> logged in.
> [2020-07-31 14:35:52,291: INFO/main] (AppInfoParser.java:117) - Kafka
> version: 5.4.0-ccs
> [2020-07-31 14:35:52,291: INFO/main] (AppInfoParser.java:118) - Kafka
> commitId: f4201a82bea68cc7
> [2020-07-31 14:35:52,291: INFO/main] (AppInfoParser.java:119) - Kafka
> startTimeMs: 1596198952287
> [2020-07-31 14:35:52,853: INFO/kafka-producer-network-thread | producer-1]
> (Metadata.java:261) - [Producer clientId=producer-1] Cluster ID: lkc-43m2m
> org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for
> alex-test-valid-data-0:5001 ms has passed since batch creation
> org.apache.kafka.common.errors.TimeoutException: Failed to allocate memory
> within the configured max blocking time 60000 ms.
> time spent blocking IO thread: 60001
> org.apache.kafka.common.errors.NetworkException: The server disconnected
> before a response was received.
> time spent blocking IO thread: 60002
> time spent blocking IO thread: 60017
> org.apache.kafka.common.errors.NetworkException: The server disconnected
> before a response was received.
> [2020-07-31 14:38:07,219: WARN/kafka-producer-network-thread | producer-1]
> (Sender.java:682) - [Producer clientId=producer-1] Received invalid metadata
> error in produce request on partition alex-test-valid-data-3 due to
> org.apache.kafka.common.errors.NetworkException: The server disconnected
> before a response was received.. Going to request metadata update now
> org.apache.kafka.common.errors.TimeoutException: Failed to allocate memory
> within the configured max blocking time 60000 ms.
> time spent blocking IO thread: 60003
> [2020-07-31 14:39:07,223: WARN/kafka-producer-network-thread | producer-1]
> (Sender.java:682) - [Producer clientId=producer-1] Received invalid metadata
> error in produce request on partition alex-test-valid-data-0 due to
> org.apache.kafka.common.errors.NetworkException: The server disconnected
> before a response was received.. Going to request metadata update now
> org.apache.kafka.common.errors.NetworkException: The server disconnected
> before a response was received.
> time spent blocking IO thread: 60002
> time spent blocking IO thread: 60001
> [2020-07-31 14:40:07,224: WARN/kafka-producer-network-thread | producer-1]
> (Sender.java:682) - [Producer clientId=producer-1] Received invalid metadata
> error in produce request on partition alex-test-valid-data-5 due to
> org.apache.kafka.common.errors.NetworkException: The server disconnected
> before a response was received.. Going to request metadata update now
> org.apache.kafka.common.errors.NetworkException: The server disconnected
> before a response was received.
> org.apache.kafka.common.errors.TimeoutException: Failed to allocate memory
> within the configured max blocking time 60000 ms.
> time spent blocking IO thread: 60001
> [2020-07-31 14:41:07,225: WARN/kafka-producer-network-thread | producer-1]
> (Sender.java:682) - [Producer clientId=producer-1] Received invalid metadata
> error in produce request on partition alex-test-valid-data-1 due to
> org.apache.kafka.common.errors.NetworkException: The server disconnected
> before a response was received.. Going to request metadata update now
> org.apache.kafka.common.errors.NetworkException: The server disconnected
> before a response was received.
> time spent blocking IO thread: 60004
> time spent blocking IO thread: 60004
> [2020-07-31 14:42:07,229: WARN/kafka-producer-network-thread | producer-1]
> (Sender.java:682) - [Producer clientId=producer-1] Received invalid metadata
> error in produce request on partition alex-test-valid-data-4 due to
> org.apache.kafka.common.errors.NetworkException: The server disconnected
> before a response was received.. Going to request metadata update now
> org.apache.kafka.common.errors.NetworkException: The server disconnected
> before a response was received.
> org.apache.kafka.common.errors.TimeoutException: Failed to allocate memory
> within the configured max blocking time 60000 ms.
> time spent blocking IO thread: 60000
> [2020-07-31 14:43:07,229: WARN/kafka-producer-network-thread | producer-1]
> (Sender.java:682) - [Producer clientId=producer-1] Received invalid metadata
> error in produce request on partition alex-test-valid-data-2 due to
> org.apache.kafka.common.errors.NetworkException: The server disconnected
> before a response was received.. Going to request metadata update now
> org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for
> alex-test-valid-data-5:422600 ms has passed since batch creation
> time spent blocking IO thread: 60003
> time spent blocking IO thread: 60001
> org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for
> alex-test-valid-data-5:422490 ms has passed since batch creation
> org.apache.kafka.common.errors.TimeoutException: Failed to allocate memory
> within the configured max blocking time 60000 ms.
> time spent blocking IO thread: 60002
> org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for
> alex-test-valid-data-5:422315 ms has passed since batch creation
> time spent blocking IO thread: 60003
> time spent blocking IO thread: 60003
> org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for
> alex-test-valid-data-5:422124 ms has passed since batch creation
> {noformat}
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
