You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/01/30 19:51:38 UTC
[GitHub] [kafka] jasonk000 opened a new pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
jasonk000 opened a new pull request #11722:
URL: https://github.com/apache/kafka/pull/11722
Hold the `deque` lock for only as long as is required to collect and make a decision in `ready()` and `drain()` loops. Once this is done, remaining work can be done without lock, so release it. This allows producers to continue appending.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jasonk000 commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
jasonk000 commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r799905219
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -448,40 +448,48 @@ public ReadyCheckResult ready(Cluster cluster, long nowMs) {
boolean exhausted = this.free.queued() > 0;
for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) {
Deque<ProducerBatch> deque = entry.getValue();
+
+ final ProducerBatch batch;
+ final long waitedTimeMs;
+ final boolean backingOff;
+ final boolean full;
+
+ // Collect as little as possible inside critical region, determine outcome after release
synchronized (deque) {
- // When producing to a large number of partitions, this path is hot and deques are often empty.
- // We check whether a batch exists first to avoid the more expensive checks whenever possible.
Review comment:
[f15fdef](https://github.com/apache/kafka/pull/11722/commits/f15fdef3241a17f86af44c6dac04ba651d629426)
Added:
```
// This loop is especially hot with large partition counts.
// We are careful to only perform the minimum required inside the
// synchronized block, as this lock is also used to synchronize producer threads
// attempting to append() to a partition/batch.
synchronized (deque) {
// Deques are often empty in this path, esp with large partition counts,
// so we exit early if we can.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma merged pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
ijuma merged pull request #11722:
URL: https://github.com/apache/kafka/pull/11722
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r799125694
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -578,41 +587,45 @@ private boolean shouldStopDrainBatchesForPartition(ProducerBatch first, TopicPar
} else {
if (shouldStopDrainBatchesForPartition(first, tp))
break;
+ }
- boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
- ProducerIdAndEpoch producerIdAndEpoch =
- transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
- ProducerBatch batch = deque.pollFirst();
- if (producerIdAndEpoch != null && !batch.hasSequence()) {
- // If the producer id/epoch of the partition do not match the latest one
- // of the producer, we update it and reset the sequence. This should be
- // only done when all its in-flight batches have completed. This is guarantee
- // in `shouldStopDrainBatchesForPartition`.
- transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
-
- // If the batch already has an assigned sequence, then we should not change the producer id and
- // sequence number, since this may introduce duplicates. In particular, the previous attempt
- // may actually have been accepted, and if we change the producer id and sequence here, this
- // attempt will also be accepted, causing a duplicate.
- //
- // Additionally, we update the next sequence number bound for the partition, and also have
- // the transaction manager track the batch so as to ensure that sequence ordering is maintained
- // even if we receive out of order responses.
- batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
- transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
- log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
- "{} being sent to partition {}", producerIdAndEpoch.producerId,
- producerIdAndEpoch.epoch, batch.baseSequence(), tp);
-
- transactionManager.addInFlightBatch(batch);
- }
- batch.close();
- size += batch.records().sizeInBytes();
- ready.add(batch);
+ // do the rest of the work by processing outside the lock
+ // close() is particularly expensive
+ batch = deque.pollFirst();
+ }
- batch.drained(now);
- }
+ boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
+ ProducerIdAndEpoch producerIdAndEpoch =
+ transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
+ if (producerIdAndEpoch != null && !batch.hasSequence()) {
+ // If the producer id/epoch of the partition do not match the latest one
+ // of the producer, we update it and reset the sequence. This should be
+ // only done when all its in-flight batches have completed. This is guarantee
+ // in `shouldStopDrainBatchesForPartition`.
+ transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
+
+ // If the batch already has an assigned sequence, then we should not change the producer id and
+ // sequence number, since this may introduce duplicates. In particular, the previous attempt
+ // may actually have been accepted, and if we change the producer id and sequence here, this
+ // attempt will also be accepted, causing a duplicate.
+ //
+ // Additionally, we update the next sequence number bound for the partition, and also have
+ // the transaction manager track the batch so as to ensure that sequence ordering is maintained
+ // even if we receive out of order responses.
+ batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
+ transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
Review comment:
Can you please test with Java 11 or newer? Looks like you tested with Java 8 which uses the slower crc32c method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jasonk000 commented on pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
jasonk000 commented on pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#issuecomment-1058761519
hi @showuon , was there anything remaining here before we merge it? thx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r801293429
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -578,41 +594,46 @@ private boolean shouldStopDrainBatchesForPartition(ProducerBatch first, TopicPar
} else {
if (shouldStopDrainBatchesForPartition(first, tp))
break;
+ }
- boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
- ProducerIdAndEpoch producerIdAndEpoch =
- transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
- ProducerBatch batch = deque.pollFirst();
- if (producerIdAndEpoch != null && !batch.hasSequence()) {
- // If the producer id/epoch of the partition do not match the latest one
- // of the producer, we update it and reset the sequence. This should be
- // only done when all its in-flight batches have completed. This is guarantee
- // in `shouldStopDrainBatchesForPartition`.
- transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
-
- // If the batch already has an assigned sequence, then we should not change the producer id and
- // sequence number, since this may introduce duplicates. In particular, the previous attempt
- // may actually have been accepted, and if we change the producer id and sequence here, this
- // attempt will also be accepted, causing a duplicate.
- //
- // Additionally, we update the next sequence number bound for the partition, and also have
- // the transaction manager track the batch so as to ensure that sequence ordering is maintained
- // even if we receive out of order responses.
- batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
- transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
- log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
- "{} being sent to partition {}", producerIdAndEpoch.producerId,
- producerIdAndEpoch.epoch, batch.baseSequence(), tp);
-
- transactionManager.addInFlightBatch(batch);
- }
- batch.close();
- size += batch.records().sizeInBytes();
- ready.add(batch);
+ batch = deque.pollFirst();
+
+ boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
+ ProducerIdAndEpoch producerIdAndEpoch =
+ transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
+ if (producerIdAndEpoch != null && !batch.hasSequence()) {
+ // If the producer id/epoch of the partition do not match the latest one
+ // of the producer, we update it and reset the sequence. This should be
+ // only done when all its in-flight batches have completed. This is guarantee
+ // in `shouldStopDrainBatchesForPartition`.
+ transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
+
+ // If the batch already has an assigned sequence, then we should not change the producer id and
+ // sequence number, since this may introduce duplicates. In particular, the previous attempt
+ // may actually have been accepted, and if we change the producer id and sequence here, this
+ // attempt will also be accepted, causing a duplicate.
+ //
+ // Additionally, we update the next sequence number bound for the partition, and also have
+ // the transaction manager track the batch so as to ensure that sequence ordering is maintained
+ // even if we receive out of order responses.
+ batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
+ transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
+ log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
+ "{} being sent to partition {}", producerIdAndEpoch.producerId,
+ producerIdAndEpoch.epoch, batch.baseSequence(), tp);
- batch.drained(now);
+ transactionManager.addInFlightBatch(batch);
}
}
+
+ // the rest of the work by processing outside the lock
+ // close() is particularly expensive
+
+ batch.close();
Review comment:
Make sense, and that's also what I've seen. Thanks for confirmation!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] artemlivshits commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
artemlivshits commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r798988451
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -448,40 +448,48 @@ public ReadyCheckResult ready(Cluster cluster, long nowMs) {
boolean exhausted = this.free.queued() > 0;
for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) {
Deque<ProducerBatch> deque = entry.getValue();
+
+ final ProducerBatch batch;
+ final long waitedTimeMs;
+ final boolean backingOff;
+ final boolean full;
+
+ // Collect as little as possible inside critical region, determine outcome after release
synchronized (deque) {
- // When producing to a large number of partitions, this path is hot and deques are often empty.
- // We check whether a batch exists first to avoid the more expensive checks whenever possible.
- ProducerBatch batch = deque.peekFirst();
- if (batch != null) {
- TopicPartition part = entry.getKey();
- Node leader = cluster.leaderFor(part);
- if (leader == null) {
- // This is a partition for which leader is not known, but messages are available to send.
- // Note that entries are currently not removed from batches when deque is empty.
- unknownLeaderTopics.add(part.topic());
- } else if (!readyNodes.contains(leader) && !isMuted(part)) {
- long waitedTimeMs = batch.waitedTimeMs(nowMs);
- boolean backingOff = batch.attempts() > 0 && waitedTimeMs < retryBackoffMs;
- long timeToWaitMs = backingOff ? retryBackoffMs : lingerMs;
- boolean full = deque.size() > 1 || batch.isFull();
- boolean expired = waitedTimeMs >= timeToWaitMs;
- boolean transactionCompleting = transactionManager != null && transactionManager.isCompleting();
- boolean sendable = full
- || expired
- || exhausted
- || closed
- || flushInProgress()
- || transactionCompleting;
- if (sendable && !backingOff) {
- readyNodes.add(leader);
- } else {
- long timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, 0);
- // Note that this results in a conservative estimate since an un-sendable partition may have
- // a leader that will later be found to have sendable data. However, this is good enough
- // since we'll just wake up and then sleep again for the remaining time.
- nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);
- }
- }
+ batch = deque.peekFirst();
Review comment:
Changes in the `ready` function LGTM
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -578,41 +587,45 @@ private boolean shouldStopDrainBatchesForPartition(ProducerBatch first, TopicPar
} else {
if (shouldStopDrainBatchesForPartition(first, tp))
break;
+ }
- boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
- ProducerIdAndEpoch producerIdAndEpoch =
- transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
- ProducerBatch batch = deque.pollFirst();
- if (producerIdAndEpoch != null && !batch.hasSequence()) {
- // If the producer id/epoch of the partition do not match the latest one
- // of the producer, we update it and reset the sequence. This should be
- // only done when all its in-flight batches have completed. This is guarantee
- // in `shouldStopDrainBatchesForPartition`.
- transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
-
- // If the batch already has an assigned sequence, then we should not change the producer id and
- // sequence number, since this may introduce duplicates. In particular, the previous attempt
- // may actually have been accepted, and if we change the producer id and sequence here, this
- // attempt will also be accepted, causing a duplicate.
- //
- // Additionally, we update the next sequence number bound for the partition, and also have
- // the transaction manager track the batch so as to ensure that sequence ordering is maintained
- // even if we receive out of order responses.
- batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
- transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
- log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
- "{} being sent to partition {}", producerIdAndEpoch.producerId,
- producerIdAndEpoch.epoch, batch.baseSequence(), tp);
-
- transactionManager.addInFlightBatch(batch);
- }
- batch.close();
- size += batch.records().sizeInBytes();
- ready.add(batch);
+ // do the rest of the work by processing outside the lock
+ // close() is particularly expensive
+ batch = deque.pollFirst();
+ }
- batch.drained(now);
- }
+ boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
+ ProducerIdAndEpoch producerIdAndEpoch =
+ transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
+ if (producerIdAndEpoch != null && !batch.hasSequence()) {
+ // If the producer id/epoch of the partition do not match the latest one
+ // of the producer, we update it and reset the sequence. This should be
+ // only done when all its in-flight batches have completed. This is guarantee
+ // in `shouldStopDrainBatchesForPartition`.
+ transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
+
+ // If the batch already has an assigned sequence, then we should not change the producer id and
+ // sequence number, since this may introduce duplicates. In particular, the previous attempt
+ // may actually have been accepted, and if we change the producer id and sequence here, this
+ // attempt will also be accepted, causing a duplicate.
+ //
+ // Additionally, we update the next sequence number bound for the partition, and also have
+ // the transaction manager track the batch so as to ensure that sequence ordering is maintained
+ // even if we receive out of order responses.
+ batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
+ transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
Review comment:
I think the expectation is that these 2 would be atomic (i.e. would be bad if one thread executed 615, then another thread executed 615 again and got the same sequence number, before the first thread got a chance to execute 616).
Also I think the expectation is that batches that are ordered one after another in the queue would get the sequence numbers in the same order (i.e. that batch that is later in the queue would get higher sequence number).
Previously these expectations were protected by the queue lock so "poll", "get sequence", "update sequence" would execute as atomic block, with this change the operations could interleave.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jasonk000 commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
jasonk000 commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r799904384
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -448,40 +448,48 @@ public ReadyCheckResult ready(Cluster cluster, long nowMs) {
boolean exhausted = this.free.queued() > 0;
for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) {
Deque<ProducerBatch> deque = entry.getValue();
+
+ final ProducerBatch batch;
+ final long waitedTimeMs;
+ final boolean backingOff;
+ final boolean full;
+
+ // Collect as little as possible inside critical region, determine outcome after release
synchronized (deque) {
- // When producing to a large number of partitions, this path is hot and deques are often empty.
- // We check whether a batch exists first to avoid the more expensive checks whenever possible.
- ProducerBatch batch = deque.peekFirst();
- if (batch != null) {
- TopicPartition part = entry.getKey();
- Node leader = cluster.leaderFor(part);
- if (leader == null) {
- // This is a partition for which leader is not known, but messages are available to send.
- // Note that entries are currently not removed from batches when deque is empty.
- unknownLeaderTopics.add(part.topic());
- } else if (!readyNodes.contains(leader) && !isMuted(part)) {
- long waitedTimeMs = batch.waitedTimeMs(nowMs);
- boolean backingOff = batch.attempts() > 0 && waitedTimeMs < retryBackoffMs;
- long timeToWaitMs = backingOff ? retryBackoffMs : lingerMs;
- boolean full = deque.size() > 1 || batch.isFull();
- boolean expired = waitedTimeMs >= timeToWaitMs;
- boolean transactionCompleting = transactionManager != null && transactionManager.isCompleting();
- boolean sendable = full
- || expired
- || exhausted
- || closed
- || flushInProgress()
- || transactionCompleting;
- if (sendable && !backingOff) {
- readyNodes.add(leader);
- } else {
- long timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, 0);
- // Note that this results in a conservative estimate since an un-sendable partition may have
- // a leader that will later be found to have sendable data. However, this is good enough
- // since we'll just wake up and then sleep again for the remaining time.
- nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);
- }
- }
+ batch = deque.peekFirst();
Review comment:
It is a small amount.
The two changes (#11721) reduce CPU of the `send()` path by about 10% on a busy app, so the impact depends on the rest of the application.
For this application, it is producing a high events/sec, so about 1.5-2% of overall system CPU is saved. For a 1000-machine cluster it will end up at ~15-20 machines.
Before
After
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r822653150
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -448,40 +448,48 @@ public ReadyCheckResult ready(Cluster cluster, long nowMs) {
boolean exhausted = this.free.queued() > 0;
for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) {
Deque<ProducerBatch> deque = entry.getValue();
+
+ final ProducerBatch batch;
+ final long waitedTimeMs;
+ final boolean backingOff;
+ final boolean full;
+
+ // Collect as little as possible inside critical region, determine outcome after release
synchronized (deque) {
- // When producing to a large number of partitions, this path is hot and deques are often empty.
- // We check whether a batch exists first to avoid the more expensive checks whenever possible.
- ProducerBatch batch = deque.peekFirst();
- if (batch != null) {
- TopicPartition part = entry.getKey();
- Node leader = cluster.leaderFor(part);
- if (leader == null) {
- // This is a partition for which leader is not known, but messages are available to send.
- // Note that entries are currently not removed from batches when deque is empty.
- unknownLeaderTopics.add(part.topic());
- } else if (!readyNodes.contains(leader) && !isMuted(part)) {
- long waitedTimeMs = batch.waitedTimeMs(nowMs);
- boolean backingOff = batch.attempts() > 0 && waitedTimeMs < retryBackoffMs;
- long timeToWaitMs = backingOff ? retryBackoffMs : lingerMs;
- boolean full = deque.size() > 1 || batch.isFull();
- boolean expired = waitedTimeMs >= timeToWaitMs;
- boolean transactionCompleting = transactionManager != null && transactionManager.isCompleting();
- boolean sendable = full
- || expired
- || exhausted
- || closed
- || flushInProgress()
- || transactionCompleting;
- if (sendable && !backingOff) {
- readyNodes.add(leader);
- } else {
- long timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, 0);
- // Note that this results in a conservative estimate since an un-sendable partition may have
- // a leader that will later be found to have sendable data. However, this is good enough
- // since we'll just wake up and then sleep again for the remaining time.
- nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);
- }
- }
+ batch = deque.peekFirst();
Review comment:
Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jasonk000 commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
jasonk000 commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r799107208
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -578,41 +587,45 @@ private boolean shouldStopDrainBatchesForPartition(ProducerBatch first, TopicPar
} else {
if (shouldStopDrainBatchesForPartition(first, tp))
break;
+ }
- boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
- ProducerIdAndEpoch producerIdAndEpoch =
- transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
- ProducerBatch batch = deque.pollFirst();
- if (producerIdAndEpoch != null && !batch.hasSequence()) {
- // If the producer id/epoch of the partition do not match the latest one
- // of the producer, we update it and reset the sequence. This should be
- // only done when all its in-flight batches have completed. This is guarantee
- // in `shouldStopDrainBatchesForPartition`.
- transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
-
- // If the batch already has an assigned sequence, then we should not change the producer id and
- // sequence number, since this may introduce duplicates. In particular, the previous attempt
- // may actually have been accepted, and if we change the producer id and sequence here, this
- // attempt will also be accepted, causing a duplicate.
- //
- // Additionally, we update the next sequence number bound for the partition, and also have
- // the transaction manager track the batch so as to ensure that sequence ordering is maintained
- // even if we receive out of order responses.
- batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
- transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
- log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
- "{} being sent to partition {}", producerIdAndEpoch.producerId,
- producerIdAndEpoch.epoch, batch.baseSequence(), tp);
-
- transactionManager.addInFlightBatch(batch);
- }
- batch.close();
- size += batch.records().sizeInBytes();
- ready.add(batch);
+ // do the rest of the work by processing outside the lock
+ // close() is particularly expensive
+ batch = deque.pollFirst();
+ }
- batch.drained(now);
- }
+ boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
+ ProducerIdAndEpoch producerIdAndEpoch =
+ transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
+ if (producerIdAndEpoch != null && !batch.hasSequence()) {
+ // If the producer id/epoch of the partition do not match the latest one
+ // of the producer, we update it and reset the sequence. This should be
+ // only done when all its in-flight batches have completed. This is guarantee
+ // in `shouldStopDrainBatchesForPartition`.
+ transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
+
+ // If the batch already has an assigned sequence, then we should not change the producer id and
+ // sequence number, since this may introduce duplicates. In particular, the previous attempt
+ // may actually have been accepted, and if we change the producer id and sequence here, this
+ // attempt will also be accepted, causing a duplicate.
+ //
+ // Additionally, we update the next sequence number bound for the partition, and also have
+ // the transaction manager track the batch so as to ensure that sequence ordering is maintained
+ // even if we receive out of order responses.
+ batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
+ transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
Review comment:
I've addressed this in 34008bf.
Below is flamegraph that `getDeque()` and `close()` consume the CPU in the `drain()` tree. I believe with both of these performed outside of the lock we are still in a good spot here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r799462063
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -448,40 +448,48 @@ public ReadyCheckResult ready(Cluster cluster, long nowMs) {
boolean exhausted = this.free.queued() > 0;
for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) {
Deque<ProducerBatch> deque = entry.getValue();
+
+ final ProducerBatch batch;
+ final long waitedTimeMs;
+ final boolean backingOff;
+ final boolean full;
+
+ // Collect as little as possible inside critical region, determine outcome after release
synchronized (deque) {
- // When producing to a large number of partitions, this path is hot and deques are often empty.
- // We check whether a batch exists first to avoid the more expensive checks whenever possible.
Review comment:
Why did we remove the information on why this is performance sensitive?
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -448,40 +448,48 @@ public ReadyCheckResult ready(Cluster cluster, long nowMs) {
boolean exhausted = this.free.queued() > 0;
for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) {
Deque<ProducerBatch> deque = entry.getValue();
+
+ final ProducerBatch batch;
+ final long waitedTimeMs;
+ final boolean backingOff;
+ final boolean full;
+
+ // Collect as little as possible inside critical region, determine outcome after release
synchronized (deque) {
- // When producing to a large number of partitions, this path is hot and deques are often empty.
- // We check whether a batch exists first to avoid the more expensive checks whenever possible.
- ProducerBatch batch = deque.peekFirst();
- if (batch != null) {
- TopicPartition part = entry.getKey();
- Node leader = cluster.leaderFor(part);
- if (leader == null) {
- // This is a partition for which leader is not known, but messages are available to send.
- // Note that entries are currently not removed from batches when deque is empty.
- unknownLeaderTopics.add(part.topic());
- } else if (!readyNodes.contains(leader) && !isMuted(part)) {
- long waitedTimeMs = batch.waitedTimeMs(nowMs);
- boolean backingOff = batch.attempts() > 0 && waitedTimeMs < retryBackoffMs;
- long timeToWaitMs = backingOff ? retryBackoffMs : lingerMs;
- boolean full = deque.size() > 1 || batch.isFull();
- boolean expired = waitedTimeMs >= timeToWaitMs;
- boolean transactionCompleting = transactionManager != null && transactionManager.isCompleting();
- boolean sendable = full
- || expired
- || exhausted
- || closed
- || flushInProgress()
- || transactionCompleting;
- if (sendable && !backingOff) {
- readyNodes.add(leader);
- } else {
- long timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, 0);
- // Note that this results in a conservative estimate since an un-sendable partition may have
- // a leader that will later be found to have sendable data. However, this is good enough
- // since we'll just wake up and then sleep again for the remaining time.
- nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);
- }
- }
+ batch = deque.peekFirst();
Review comment:
Out of curiosity, do these make much of a difference? We can probably keep them, but it seems like the big problem was the `close` call being inside the lock (in the other method).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jasonk000 commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
jasonk000 commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r801259547
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -578,41 +594,46 @@ private boolean shouldStopDrainBatchesForPartition(ProducerBatch first, TopicPar
} else {
if (shouldStopDrainBatchesForPartition(first, tp))
break;
+ }
- boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
- ProducerIdAndEpoch producerIdAndEpoch =
- transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
- ProducerBatch batch = deque.pollFirst();
- if (producerIdAndEpoch != null && !batch.hasSequence()) {
- // If the producer id/epoch of the partition do not match the latest one
- // of the producer, we update it and reset the sequence. This should be
- // only done when all its in-flight batches have completed. This is guarantee
- // in `shouldStopDrainBatchesForPartition`.
- transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
-
- // If the batch already has an assigned sequence, then we should not change the producer id and
- // sequence number, since this may introduce duplicates. In particular, the previous attempt
- // may actually have been accepted, and if we change the producer id and sequence here, this
- // attempt will also be accepted, causing a duplicate.
- //
- // Additionally, we update the next sequence number bound for the partition, and also have
- // the transaction manager track the batch so as to ensure that sequence ordering is maintained
- // even if we receive out of order responses.
- batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
- transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
- log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
- "{} being sent to partition {}", producerIdAndEpoch.producerId,
- producerIdAndEpoch.epoch, batch.baseSequence(), tp);
-
- transactionManager.addInFlightBatch(batch);
- }
- batch.close();
- size += batch.records().sizeInBytes();
- ready.add(batch);
+ batch = deque.pollFirst();
+
+ boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
+ ProducerIdAndEpoch producerIdAndEpoch =
+ transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
+ if (producerIdAndEpoch != null && !batch.hasSequence()) {
+ // If the producer id/epoch of the partition do not match the latest one
+ // of the producer, we update it and reset the sequence. This should be
+ // only done when all its in-flight batches have completed. This is guarantee
+ // in `shouldStopDrainBatchesForPartition`.
+ transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
+
+ // If the batch already has an assigned sequence, then we should not change the producer id and
+ // sequence number, since this may introduce duplicates. In particular, the previous attempt
+ // may actually have been accepted, and if we change the producer id and sequence here, this
+ // attempt will also be accepted, causing a duplicate.
+ //
+ // Additionally, we update the next sequence number bound for the partition, and also have
+ // the transaction manager track the batch so as to ensure that sequence ordering is maintained
+ // even if we receive out of order responses.
+ batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
+ transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
+ log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
+ "{} being sent to partition {}", producerIdAndEpoch.producerId,
+ producerIdAndEpoch.epoch, batch.baseSequence(), tp);
- batch.drained(now);
+ transactionManager.addInFlightBatch(batch);
}
}
+
+ // the rest of the work by processing outside the lock
+ // close() is particularly expensive
+
+ batch.close();
Review comment:
Since we lock on deque, the batch is either "in" the deque and append/etc will work on it in synchronized block, or it is removed from deque, and will no longer be available for other threads to do any work on it. From my reading, Sender only collects batches via the drain and expiry path, both of which acquire the lock before removing the element from the deque. This implies that the removal from batch is atomic, whether that is drained or expired out, it can only be one. The Sender re-enqueue works on batches that were previously drained as well. Nothing stands out at me as a possible area for leakage but I'm happy to be corrected and more eyes is always better for this sort of thing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r799125694
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -578,41 +587,45 @@ private boolean shouldStopDrainBatchesForPartition(ProducerBatch first, TopicPar
} else {
if (shouldStopDrainBatchesForPartition(first, tp))
break;
+ }
- boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
- ProducerIdAndEpoch producerIdAndEpoch =
- transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
- ProducerBatch batch = deque.pollFirst();
- if (producerIdAndEpoch != null && !batch.hasSequence()) {
- // If the producer id/epoch of the partition do not match the latest one
- // of the producer, we update it and reset the sequence. This should be
- // only done when all its in-flight batches have completed. This is guarantee
- // in `shouldStopDrainBatchesForPartition`.
- transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
-
- // If the batch already has an assigned sequence, then we should not change the producer id and
- // sequence number, since this may introduce duplicates. In particular, the previous attempt
- // may actually have been accepted, and if we change the producer id and sequence here, this
- // attempt will also be accepted, causing a duplicate.
- //
- // Additionally, we update the next sequence number bound for the partition, and also have
- // the transaction manager track the batch so as to ensure that sequence ordering is maintained
- // even if we receive out of order responses.
- batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
- transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
- log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
- "{} being sent to partition {}", producerIdAndEpoch.producerId,
- producerIdAndEpoch.epoch, batch.baseSequence(), tp);
-
- transactionManager.addInFlightBatch(batch);
- }
- batch.close();
- size += batch.records().sizeInBytes();
- ready.add(batch);
+ // do the rest of the work by processing outside the lock
+ // close() is particularly expensive
+ batch = deque.pollFirst();
+ }
- batch.drained(now);
- }
+ boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
+ ProducerIdAndEpoch producerIdAndEpoch =
+ transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
+ if (producerIdAndEpoch != null && !batch.hasSequence()) {
+ // If the producer id/epoch of the partition do not match the latest one
+ // of the producer, we update it and reset the sequence. This should be
+ // only done when all its in-flight batches have completed. This is guarantee
+ // in `shouldStopDrainBatchesForPartition`.
+ transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
+
+ // If the batch already has an assigned sequence, then we should not change the producer id and
+ // sequence number, since this may introduce duplicates. In particular, the previous attempt
+ // may actually have been accepted, and if we change the producer id and sequence here, this
+ // attempt will also be accepted, causing a duplicate.
+ //
+ // Additionally, we update the next sequence number bound for the partition, and also have
+ // the transaction manager track the batch so as to ensure that sequence ordering is maintained
+ // even if we receive out of order responses.
+ batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
+ transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
Review comment:
Can you please test with Java 11 or newer? Looks like you tested with Java 8 which used the slower crc32c method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r799428683
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -578,41 +587,45 @@ private boolean shouldStopDrainBatchesForPartition(ProducerBatch first, TopicPar
} else {
if (shouldStopDrainBatchesForPartition(first, tp))
break;
+ }
- boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
- ProducerIdAndEpoch producerIdAndEpoch =
- transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
- ProducerBatch batch = deque.pollFirst();
- if (producerIdAndEpoch != null && !batch.hasSequence()) {
- // If the producer id/epoch of the partition do not match the latest one
- // of the producer, we update it and reset the sequence. This should be
- // only done when all its in-flight batches have completed. This is guarantee
- // in `shouldStopDrainBatchesForPartition`.
- transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
-
- // If the batch already has an assigned sequence, then we should not change the producer id and
- // sequence number, since this may introduce duplicates. In particular, the previous attempt
- // may actually have been accepted, and if we change the producer id and sequence here, this
- // attempt will also be accepted, causing a duplicate.
- //
- // Additionally, we update the next sequence number bound for the partition, and also have
- // the transaction manager track the batch so as to ensure that sequence ordering is maintained
- // even if we receive out of order responses.
- batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
- transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
- log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
- "{} being sent to partition {}", producerIdAndEpoch.producerId,
- producerIdAndEpoch.epoch, batch.baseSequence(), tp);
-
- transactionManager.addInFlightBatch(batch);
- }
- batch.close();
- size += batch.records().sizeInBytes();
- ready.add(batch);
+ // do the rest of the work by processing outside the lock
+ // close() is particularly expensive
+ batch = deque.pollFirst();
+ }
- batch.drained(now);
- }
+ boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
+ ProducerIdAndEpoch producerIdAndEpoch =
+ transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
+ if (producerIdAndEpoch != null && !batch.hasSequence()) {
+ // If the producer id/epoch of the partition do not match the latest one
+ // of the producer, we update it and reset the sequence. This should be
+ // only done when all its in-flight batches have completed. This is guarantee
+ // in `shouldStopDrainBatchesForPartition`.
+ transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
+
+ // If the batch already has an assigned sequence, then we should not change the producer id and
+ // sequence number, since this may introduce duplicates. In particular, the previous attempt
+ // may actually have been accepted, and if we change the producer id and sequence here, this
+ // attempt will also be accepted, causing a duplicate.
+ //
+ // Additionally, we update the next sequence number bound for the partition, and also have
+ // the transaction manager track the batch so as to ensure that sequence ordering is maintained
+ // even if we receive out of order responses.
+ batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
+ transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
Review comment:
No, not a blocker.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r800322496
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -578,41 +594,46 @@ private boolean shouldStopDrainBatchesForPartition(ProducerBatch first, TopicPar
} else {
if (shouldStopDrainBatchesForPartition(first, tp))
break;
+ }
- boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
- ProducerIdAndEpoch producerIdAndEpoch =
- transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
- ProducerBatch batch = deque.pollFirst();
- if (producerIdAndEpoch != null && !batch.hasSequence()) {
- // If the producer id/epoch of the partition do not match the latest one
- // of the producer, we update it and reset the sequence. This should be
- // only done when all its in-flight batches have completed. This is guarantee
- // in `shouldStopDrainBatchesForPartition`.
- transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
-
- // If the batch already has an assigned sequence, then we should not change the producer id and
- // sequence number, since this may introduce duplicates. In particular, the previous attempt
- // may actually have been accepted, and if we change the producer id and sequence here, this
- // attempt will also be accepted, causing a duplicate.
- //
- // Additionally, we update the next sequence number bound for the partition, and also have
- // the transaction manager track the batch so as to ensure that sequence ordering is maintained
- // even if we receive out of order responses.
- batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
- transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
- log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
- "{} being sent to partition {}", producerIdAndEpoch.producerId,
- producerIdAndEpoch.epoch, batch.baseSequence(), tp);
-
- transactionManager.addInFlightBatch(batch);
- }
- batch.close();
- size += batch.records().sizeInBytes();
- ready.add(batch);
+ batch = deque.pollFirst();
+
+ boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
+ ProducerIdAndEpoch producerIdAndEpoch =
+ transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
+ if (producerIdAndEpoch != null && !batch.hasSequence()) {
+ // If the producer id/epoch of the partition do not match the latest one
+ // of the producer, we update it and reset the sequence. This should be
+ // only done when all its in-flight batches have completed. This is guarantee
+ // in `shouldStopDrainBatchesForPartition`.
+ transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
+
+ // If the batch already has an assigned sequence, then we should not change the producer id and
+ // sequence number, since this may introduce duplicates. In particular, the previous attempt
+ // may actually have been accepted, and if we change the producer id and sequence here, this
+ // attempt will also be accepted, causing a duplicate.
+ //
+ // Additionally, we update the next sequence number bound for the partition, and also have
+ // the transaction manager track the batch so as to ensure that sequence ordering is maintained
+ // even if we receive out of order responses.
+ batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
+ transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
+ log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
+ "{} being sent to partition {}", producerIdAndEpoch.producerId,
+ producerIdAndEpoch.epoch, batch.baseSequence(), tp);
- batch.drained(now);
+ transactionManager.addInFlightBatch(batch);
}
}
+
+ // the rest of the work by processing outside the lock
+ // close() is particularly expensive
+
+ batch.close();
Review comment:
If we move the `batch.close()` out of synchronized scope, is it possible that other thread might take the batch as "alive" and do other operation before we close it completely?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#issuecomment-1058762845
@ijuma @artemlivshits , do you want to have another look at this PR? Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jasonk000 commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
jasonk000 commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r799030101
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -578,41 +587,45 @@ private boolean shouldStopDrainBatchesForPartition(ProducerBatch first, TopicPar
} else {
if (shouldStopDrainBatchesForPartition(first, tp))
break;
+ }
- boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
- ProducerIdAndEpoch producerIdAndEpoch =
- transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
- ProducerBatch batch = deque.pollFirst();
- if (producerIdAndEpoch != null && !batch.hasSequence()) {
- // If the producer id/epoch of the partition do not match the latest one
- // of the producer, we update it and reset the sequence. This should be
- // only done when all its in-flight batches have completed. This is guarantee
- // in `shouldStopDrainBatchesForPartition`.
- transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
-
- // If the batch already has an assigned sequence, then we should not change the producer id and
- // sequence number, since this may introduce duplicates. In particular, the previous attempt
- // may actually have been accepted, and if we change the producer id and sequence here, this
- // attempt will also be accepted, causing a duplicate.
- //
- // Additionally, we update the next sequence number bound for the partition, and also have
- // the transaction manager track the batch so as to ensure that sequence ordering is maintained
- // even if we receive out of order responses.
- batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
- transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
- log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
- "{} being sent to partition {}", producerIdAndEpoch.producerId,
- producerIdAndEpoch.epoch, batch.baseSequence(), tp);
-
- transactionManager.addInFlightBatch(batch);
- }
- batch.close();
- size += batch.records().sizeInBytes();
- ready.add(batch);
+ // do the rest of the work by processing outside the lock
+ // close() is particularly expensive
+ batch = deque.pollFirst();
+ }
- batch.drained(now);
- }
+ boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
+ ProducerIdAndEpoch producerIdAndEpoch =
+ transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
+ if (producerIdAndEpoch != null && !batch.hasSequence()) {
+ // If the producer id/epoch of the partition do not match the latest one
+ // of the producer, we update it and reset the sequence. This should be
+ // only done when all its in-flight batches have completed. This is guarantee
+ // in `shouldStopDrainBatchesForPartition`.
+ transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
+
+ // If the batch already has an assigned sequence, then we should not change the producer id and
+ // sequence number, since this may introduce duplicates. In particular, the previous attempt
+ // may actually have been accepted, and if we change the producer id and sequence here, this
+ // attempt will also be accepted, causing a duplicate.
+ //
+ // Additionally, we update the next sequence number bound for the partition, and also have
+ // the transaction manager track the batch so as to ensure that sequence ordering is maintained
+ // even if we receive out of order responses.
+ batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
+ transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
Review comment:
OK, this makes sense. I agree it makes sense to push this back into the synchronized block; the more important part for reducing lock duration is to move the `close()` outside of the block. Would you agree with this approach?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] artemlivshits commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
artemlivshits commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r799117854
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -578,41 +587,45 @@ private boolean shouldStopDrainBatchesForPartition(ProducerBatch first, TopicPar
} else {
if (shouldStopDrainBatchesForPartition(first, tp))
break;
+ }
- boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
- ProducerIdAndEpoch producerIdAndEpoch =
- transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
- ProducerBatch batch = deque.pollFirst();
- if (producerIdAndEpoch != null && !batch.hasSequence()) {
- // If the producer id/epoch of the partition do not match the latest one
- // of the producer, we update it and reset the sequence. This should be
- // only done when all its in-flight batches have completed. This is guarantee
- // in `shouldStopDrainBatchesForPartition`.
- transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
-
- // If the batch already has an assigned sequence, then we should not change the producer id and
- // sequence number, since this may introduce duplicates. In particular, the previous attempt
- // may actually have been accepted, and if we change the producer id and sequence here, this
- // attempt will also be accepted, causing a duplicate.
- //
- // Additionally, we update the next sequence number bound for the partition, and also have
- // the transaction manager track the batch so as to ensure that sequence ordering is maintained
- // even if we receive out of order responses.
- batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
- transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
- log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
- "{} being sent to partition {}", producerIdAndEpoch.producerId,
- producerIdAndEpoch.epoch, batch.baseSequence(), tp);
-
- transactionManager.addInFlightBatch(batch);
- }
- batch.close();
- size += batch.records().sizeInBytes();
- ready.add(batch);
+ // do the rest of the work by processing outside the lock
+ // close() is particularly expensive
+ batch = deque.pollFirst();
+ }
- batch.drained(now);
- }
+ boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
+ ProducerIdAndEpoch producerIdAndEpoch =
+ transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
+ if (producerIdAndEpoch != null && !batch.hasSequence()) {
+ // If the producer id/epoch of the partition do not match the latest one
+ // of the producer, we update it and reset the sequence. This should be
+ // only done when all its in-flight batches have completed. This is guarantee
+ // in `shouldStopDrainBatchesForPartition`.
+ transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
+
+ // If the batch already has an assigned sequence, then we should not change the producer id and
+ // sequence number, since this may introduce duplicates. In particular, the previous attempt
+ // may actually have been accepted, and if we change the producer id and sequence here, this
+ // attempt will also be accepted, causing a duplicate.
+ //
+ // Additionally, we update the next sequence number bound for the partition, and also have
+ // the transaction manager track the batch so as to ensure that sequence ordering is maintained
+ // even if we receive out of order responses.
+ batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
+ transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
Review comment:
Moving `close` outside of locked scope LGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jasonk000 commented on pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
jasonk000 commented on pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#issuecomment-1026000556
This application has a single producer thread, a high `send()` rate. The change reduces spinlock CPU cycles from 14.6% to 2.5% of the `send()` path, or more clearly a 12.1% improvement in efficiency for the `send()` path by reducing the duration of contention events with the network thread.
Before, with `*complete_monitor*` highlighted:
After, with `*complete_monitor*` highlighted:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#issuecomment-1025708768
Thanks for the PR. Do you have some results to share regarding the impact this optimization has?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jasonk000 commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
jasonk000 commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r799138886
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -578,41 +587,45 @@ private boolean shouldStopDrainBatchesForPartition(ProducerBatch first, TopicPar
} else {
if (shouldStopDrainBatchesForPartition(first, tp))
break;
+ }
- boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
- ProducerIdAndEpoch producerIdAndEpoch =
- transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
- ProducerBatch batch = deque.pollFirst();
- if (producerIdAndEpoch != null && !batch.hasSequence()) {
- // If the producer id/epoch of the partition do not match the latest one
- // of the producer, we update it and reset the sequence. This should be
- // only done when all its in-flight batches have completed. This is guarantee
- // in `shouldStopDrainBatchesForPartition`.
- transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
-
- // If the batch already has an assigned sequence, then we should not change the producer id and
- // sequence number, since this may introduce duplicates. In particular, the previous attempt
- // may actually have been accepted, and if we change the producer id and sequence here, this
- // attempt will also be accepted, causing a duplicate.
- //
- // Additionally, we update the next sequence number bound for the partition, and also have
- // the transaction manager track the batch so as to ensure that sequence ordering is maintained
- // even if we receive out of order responses.
- batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
- transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
- log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
- "{} being sent to partition {}", producerIdAndEpoch.producerId,
- producerIdAndEpoch.epoch, batch.baseSequence(), tp);
-
- transactionManager.addInFlightBatch(batch);
- }
- batch.close();
- size += batch.records().sizeInBytes();
- ready.add(batch);
+ // do the rest of the work by processing outside the lock
+ // close() is particularly expensive
+ batch = deque.pollFirst();
+ }
- batch.drained(now);
- }
+ boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
+ ProducerIdAndEpoch producerIdAndEpoch =
+ transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
+ if (producerIdAndEpoch != null && !batch.hasSequence()) {
+ // If the producer id/epoch of the partition do not match the latest one
+ // of the producer, we update it and reset the sequence. This should be
+ // only done when all its in-flight batches have completed. This is guarantee
+ // in `shouldStopDrainBatchesForPartition`.
+ transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
+
+ // If the batch already has an assigned sequence, then we should not change the producer id and
+ // sequence number, since this may introduce duplicates. In particular, the previous attempt
+ // may actually have been accepted, and if we change the producer id and sequence here, this
+ // attempt will also be accepted, causing a duplicate.
+ //
+ // Additionally, we update the next sequence number bound for the partition, and also have
+ // the transaction manager track the batch so as to ensure that sequence ordering is maintained
+ // even if we receive out of order responses.
+ batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
+ transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
Review comment:
Correct, this application is jdk8. I'll have to find a jdk11 app to compare, it could take a while as this app is not ready for it. Is a jdk11 test a blocker?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r799828526
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -448,40 +448,48 @@ public ReadyCheckResult ready(Cluster cluster, long nowMs) {
boolean exhausted = this.free.queued() > 0;
for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) {
Deque<ProducerBatch> deque = entry.getValue();
+
+ final ProducerBatch batch;
+ final long waitedTimeMs;
+ final boolean backingOff;
+ final boolean full;
+
+ // Collect as little as possible inside critical region, determine outcome after release
synchronized (deque) {
- // When producing to a large number of partitions, this path is hot and deques are often empty.
- // We check whether a batch exists first to avoid the more expensive checks whenever possible.
Review comment:
I think the useful context that was removed is:
> When producing to a large number of partitions, this path is hot and deques are often empty.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r800338273
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -578,41 +594,46 @@ private boolean shouldStopDrainBatchesForPartition(ProducerBatch first, TopicPar
} else {
if (shouldStopDrainBatchesForPartition(first, tp))
break;
+ }
- boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
- ProducerIdAndEpoch producerIdAndEpoch =
- transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
- ProducerBatch batch = deque.pollFirst();
- if (producerIdAndEpoch != null && !batch.hasSequence()) {
- // If the producer id/epoch of the partition do not match the latest one
- // of the producer, we update it and reset the sequence. This should be
- // only done when all its in-flight batches have completed. This is guarantee
- // in `shouldStopDrainBatchesForPartition`.
- transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
-
- // If the batch already has an assigned sequence, then we should not change the producer id and
- // sequence number, since this may introduce duplicates. In particular, the previous attempt
- // may actually have been accepted, and if we change the producer id and sequence here, this
- // attempt will also be accepted, causing a duplicate.
- //
- // Additionally, we update the next sequence number bound for the partition, and also have
- // the transaction manager track the batch so as to ensure that sequence ordering is maintained
- // even if we receive out of order responses.
- batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
- transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
- log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
- "{} being sent to partition {}", producerIdAndEpoch.producerId,
- producerIdAndEpoch.epoch, batch.baseSequence(), tp);
-
- transactionManager.addInFlightBatch(batch);
- }
- batch.close();
- size += batch.records().sizeInBytes();
- ready.add(batch);
+ batch = deque.pollFirst();
+
+ boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
+ ProducerIdAndEpoch producerIdAndEpoch =
+ transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
+ if (producerIdAndEpoch != null && !batch.hasSequence()) {
+ // If the producer id/epoch of the partition do not match the latest one
+ // of the producer, we update it and reset the sequence. This should be
+ // only done when all its in-flight batches have completed. This is guarantee
+ // in `shouldStopDrainBatchesForPartition`.
+ transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);
+
+ // If the batch already has an assigned sequence, then we should not change the producer id and
+ // sequence number, since this may introduce duplicates. In particular, the previous attempt
+ // may actually have been accepted, and if we change the producer id and sequence here, this
+ // attempt will also be accepted, causing a duplicate.
+ //
+ // Additionally, we update the next sequence number bound for the partition, and also have
+ // the transaction manager track the batch so as to ensure that sequence ordering is maintained
+ // even if we receive out of order responses.
+ batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
+ transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
+ log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
+ "{} being sent to partition {}", producerIdAndEpoch.producerId,
+ producerIdAndEpoch.epoch, batch.baseSequence(), tp);
- batch.drained(now);
+ transactionManager.addInFlightBatch(batch);
}
}
+
+ // the rest of the work by processing outside the lock
+ // close() is particularly expensive
+
+ batch.close();
Review comment:
Looks like it won't happen since we only lock on deque object, but just want to confirm, to make sure it won't break anything.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jasonk000 commented on a change in pull request #11722: KAFKA-13630: reduce amount of time that producer network thread holds batch queue lock
Posted by GitBox <gi...@apache.org>.
jasonk000 commented on a change in pull request #11722:
URL: https://github.com/apache/kafka/pull/11722#discussion_r799796771
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
##########
@@ -448,40 +448,48 @@ public ReadyCheckResult ready(Cluster cluster, long nowMs) {
boolean exhausted = this.free.queued() > 0;
for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) {
Deque<ProducerBatch> deque = entry.getValue();
+
+ final ProducerBatch batch;
+ final long waitedTimeMs;
+ final boolean backingOff;
+ final boolean full;
+
+ // Collect as little as possible inside critical region, determine outcome after release
synchronized (deque) {
- // When producing to a large number of partitions, this path is hot and deques are often empty.
- // We check whether a batch exists first to avoid the more expensive checks whenever possible.
Review comment:
I attempted to capture the sentiment with this comment, given the re-flowed logic:
```
// Collect as little as possible inside critical region, determine outcome after release
```
I'll add something more descriptive.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org