You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "gongxuanzhang (via GitHub)" <gi...@apache.org> on 2025/04/16 09:49:51 UTC
[PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
gongxuanzhang opened a new pull request, #19489:
URL: https://github.com/apache/kafka/pull/19489
issue see https://issues.apache.org/jira/browse/KAFKA-17862
đ Problem Summary
When an expired batch is still part of an in-flight request, we prematurely release the ByteBuffer back to the BufferPool. This leads to two critical issues:
1. Expiration does not prevent the in-flight request from being sent.
2. The expired batchâs ByteBuffer is deallocate to the pool too early. It may be re-allocated for another producer batch while still being referenced by the in-flight request, potentially causing data corruption.
We can tolerate Issue 1, but Issue 2 is critical â we cannot allow it to happen.
Therefore, we remove the expiration handling of ProducerBatch before send, and instead defer the ByteBuffer deallocation to the response handling logic.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3182080629
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2612164696
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:
##########
@@ -1602,13 +1602,34 @@ private MemoryRecords trimInvalidBytes(MemoryRecords records, LogAppendInfo info
if (validBytes == records.sizeInBytes()) {
return records;
} else {
- // trim invalid bytes
+ // Duplicate the original buffer for trimming and logging purposes.
ByteBuffer validByteBuffer = records.buffer().duplicate();
+
+ // Log detailed information about trimmed bytes
+ validByteBuffer.position(validBytes);
+ byte[] invalidBytes = new byte[records.sizeInBytes() - validBytes];
+ validByteBuffer.get(invalidBytes);
+ String invalidBytesHex = bytesToHex(invalidBytes);
+
+ logger.warn("Trimming invalid bytes from message set for partition {}. Original size: {} bytes, valid bytes: {}, trimmed bytes: {}. " +
Review Comment:
@junrao, is it sufficient to pass the `AppendOrigin` from `append()` to `trimInvalidBytes()` and then check if it's equal to `AppendOrigin.CLIENT`?
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2612103020
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
- List<ProducerBatch> expiredBatches = new ArrayList<>();
-
- for (Iterator<Map.Entry<TopicPartition, List<ProducerBatch>>> batchIt = inFlightBatches.entrySet().iterator(); batchIt.hasNext();) {
- Map.Entry<TopicPartition, List<ProducerBatch>> entry = batchIt.next();
- List<ProducerBatch> partitionInFlightBatches = entry.getValue();
- if (partitionInFlightBatches != null) {
- Iterator<ProducerBatch> iter = partitionInFlightBatches.iterator();
- while (iter.hasNext()) {
- ProducerBatch batch = iter.next();
- if (batch.hasReachedDeliveryTimeout(accumulator.getDeliveryTimeoutMs(), now)) {
- iter.remove();
- // expireBatches is called in Sender.sendProducerData, before client.poll.
- // The !batch.isDone() invariant should always hold. An IllegalStateException
- // exception will be thrown if the invariant is violated.
- if (!batch.isDone()) {
- expiredBatches.add(batch);
- } else {
- throw new IllegalStateException(batch.topicPartition + " batch created at " +
- batch.createdMs + " gets unexpected final state " + batch.finalState());
- }
- } else {
- accumulator.maybeUpdateNextBatchExpiryTime(batch);
- break;
- }
- }
- if (partitionInFlightBatches.isEmpty()) {
- batchIt.remove();
- }
- }
- }
- return expiredBatches;
- }
-
Review Comment:
CI in place, resolving comment.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3661687614
@gongxuanzhangâthank you very much for finding a precision fix for the buffer corruption issue!
In terms of the Kafka 4.2.0 release cycle, the code freeze deadline has passed and we're focusing on blockers. KAFKA-17862 is marked as a blocker for 4.2.0, so let @junrao, @chia7712, or me know if you need some help resolving the issue.
Thanks for the work so far đđđđ
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2644690445
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
That is unavoidable since a request timeout does not mean the request failed either.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3535500718
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2596837495
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java:
##########
@@ -48,7 +48,7 @@ public class BufferPool {
private final long totalMemory;
private final int poolableSize;
- private final ReentrantLock lock;
+ protected final ReentrantLock lock;
Review Comment:
At that time, I planned to create a new BufferPool, which would have required using this lock. But now I have a better approach, so itâs no longer needed.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3629655513
There's a weird error that's failing both the [JDK 17](https://github.com/apache/kafka/actions/runs/20018321623/job/57400884392?pr=19489) and [JDK 25](https://github.com/apache/kafka/actions/runs/20018321623/job/57400884363?pr=19489) builds during the JUnit XML parsing step:
```
. . .
Parsing file: build/junit-xml/clients/25-noflaky-nonew/TEST-org.apache.kafka.common.record.ControlRecordTypeTest.xml, module: clients, job: 25-noflaky-nonew
Parsing file: build/junit-xml/clients/25-noflaky-nonew/TEST-org.apache.kafka.common.requests.TxnOffsetCommitRequestTest.xml, module: clients, job: 25-noflaky-nonew
Parsing file: build/junit-xml/clients/25-noflaky-nonew/TEST-org.apache.kafka.common.record.LegacyRecordTest.xml, module: clients, job: 25-noflaky-nonew
Parsing file: build/junit-xml/clients/25-noflaky-nonew/TEST-org.apache.kafka.common.security.oauthbearer.ClaimValidationUtilsTest.xml, module: clients, job: 25-noflaky-nonew
Parsing file: build/junit-xml/clients/25-noflaky-nonew/TEST-org.apache.kafka.clients.consumer.internals.ApplicationEventHandlerTest.xml, module: clients, job: 25-noflaky-nonew
Traceback (most recent call last):
File "/home/runner/work/kafka/kafka/.github/scripts/junit.py", line 278, in <module>
for suite in parse_report(workspace_path, report, fp):
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/runner/work/kafka/kafka/.github/scripts/junit.py", line 149, in parse_report
for (event, elem) in xml.etree.ElementTree.iterparse(fp, events=["start", "end"]):
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/hostedtoolcache/Python/3.12.12/x64/lib/python3.12/xml/etree/ElementTree.py", line 1242, in iterator
root = pullparser._close_and_return_root()
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/hostedtoolcache/Python/3.12.12/x64/lib/python3.12/xml/etree/ElementTree.py", line 1290, in _close_and_return_root
root = self._parser.close()
^^^^^^^^^^^^^^^^^^^^
xml.etree.ElementTree.ParseError: no element found: line 1, column 0
```
I'll look around and see if I can get more detail.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2632744006
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2686,14 +2701,14 @@ public void testResetNextBatchExpiry() throws Exception {
inOrder.verify(client, atLeastOnce()).newClientRequest(anyString(), any(), anyLong(), anyBoolean(), anyInt(), any());
inOrder.verify(client, atLeastOnce()).send(any(), anyLong());
inOrder.verify(client).poll(eq(0L), anyLong());
- inOrder.verify(client).poll(eq(accumulator.getDeliveryTimeoutMs()), anyLong());
+ inOrder.verify(client).poll(geq(Long.MAX_VALUE - System.currentTimeMillis()), anyLong());
Review Comment:
We need to use time, instead of System time. We also need to save the time before the 2nd runOnce() call.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3676038611
> Any recommendations on fixing the tests in TransactionManagerTest due to the changes in this PR?
It seems the CI has passed, but I'm not sure I understand the situation fully yet. Did we figure something out?
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2642070992
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
Hmm, this seems like an undesirable regression. I haven't followed the PR in detail - why do we have to regress in this way?
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2590017754
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -403,9 +367,7 @@ private long sendProducerData(long now) {
}
accumulator.resetNextBatchExpiryTime();
- List<ProducerBatch> expiredInflightBatches = getExpiredInflightBatches(now);
Review Comment:
One impact of this change is that now a record could take longer than delivery time to complete. One way to address this is to set the timeout for client.newClientRequest as the lower of the request timeout and the remaining delivery time since the batch creation time.
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerIntegrationTest.java:
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.producer;
+
+import org.apache.kafka.clients.ApiVersions;
+import org.apache.kafka.clients.ClientUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.KafkaClient;
+import org.apache.kafka.clients.producer.internals.BufferPool;
+import org.apache.kafka.clients.producer.internals.ProducerInterceptors;
+import org.apache.kafka.clients.producer.internals.ProducerMetadata;
+import org.apache.kafka.clients.producer.internals.ProducerMetrics;
+import org.apache.kafka.clients.producer.internals.RecordAccumulator;
+import org.apache.kafka.clients.producer.internals.Sender;
+import org.apache.kafka.common.compress.NoCompression;
+import org.apache.kafka.common.internals.ClusterResourceListeners;
+import org.apache.kafka.common.internals.Plugin;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.MetricsReporter;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.TestUtils;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
+
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+
+
+public class ProducerIntegrationTest {
+
+ @ClusterTest(serverProperties = {
+ @ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1"),
+ })
+ public void testInFlightBatchShouldNotBeCorrupted(ClusterInstance cluster) throws InterruptedException,
+ ExecutionException {
+ String topic = "test-topic";
+ cluster.createTopic("test-topic", 1, (short) 1);
+ try (var producer = expireProducer(cluster)) {
+ producer.send(new ProducerRecord<>(topic, "key".getBytes(), "value".getBytes())).get();
+ }
+ try (var consumer = cluster.consumer()) {
+ consumer.subscribe(List.of(topic));
+ TestUtils.waitForCondition(() -> consumer.poll(Duration.ofSeconds(1)).count() == 1, 5000, "failed to poll data");
+ }
+
+ }
+
+
+ @SuppressWarnings({"unchecked", "this-escape"})
+ private Producer<byte[], byte[]> expireProducer(ClusterInstance cluster) {
+ Map<String, Object> config = Map.of(
+ ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName(),
+ ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName(),
+ ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers(),
+ ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false,
+ ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 2000,
+ ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 1500
+ );
+ return new EvilKafkaProducerBuilder().build(config);
+ }
+
+ static class EvilKafkaProducerBuilder {
+
+ Serializer<byte[]> serializer = new ByteArraySerializer();
+ ApiVersions apiVersions = new ApiVersions();
+ LogContext logContext = new LogContext("[expire Producer test ]");
+ Metrics metrics = new Metrics(Time.SYSTEM);
+
+ String clientId;
+ String transactionalId;
+ ProducerConfig config;
+ ProducerMetadata metadata;
+ RecordAccumulator accumulator;
+ Partitioner partitioner;
+ Sender sender;
+ ProducerInterceptors<String, String> interceptors;
+
+ @SuppressWarnings({"unchecked", "this-escape"})
+ Producer<byte[], byte[]> build(Map<String, Object> configs) {
+ this.config = new ProducerConfig(ProducerConfig.appendSerializerToConfig(configs, null, null));
Review Comment:
Could we be consistent with the usage of `this.`? We don't use `this.` for other instance variables. Ditto in a few other places.
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerIntegrationTest.java:
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.producer;
+
+import org.apache.kafka.clients.ApiVersions;
+import org.apache.kafka.clients.ClientUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.KafkaClient;
+import org.apache.kafka.clients.producer.internals.BufferPool;
+import org.apache.kafka.clients.producer.internals.ProducerInterceptors;
+import org.apache.kafka.clients.producer.internals.ProducerMetadata;
+import org.apache.kafka.clients.producer.internals.ProducerMetrics;
+import org.apache.kafka.clients.producer.internals.RecordAccumulator;
+import org.apache.kafka.clients.producer.internals.Sender;
+import org.apache.kafka.common.compress.NoCompression;
+import org.apache.kafka.common.internals.ClusterResourceListeners;
+import org.apache.kafka.common.internals.Plugin;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.MetricsReporter;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.TestUtils;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
+
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+
+
+public class ProducerIntegrationTest {
+
+ @ClusterTest(serverProperties = {
+ @ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1"),
+ })
+ public void testInFlightBatchShouldNotBeCorrupted(ClusterInstance cluster) throws InterruptedException,
+ ExecutionException {
+ String topic = "test-topic";
+ cluster.createTopic("test-topic", 1, (short) 1);
+ try (var producer = expireProducer(cluster)) {
+ producer.send(new ProducerRecord<>(topic, "key".getBytes(), "value".getBytes())).get();
+ }
+ try (var consumer = cluster.consumer()) {
+ consumer.subscribe(List.of(topic));
+ TestUtils.waitForCondition(() -> consumer.poll(Duration.ofSeconds(1)).count() == 1, 5000, "failed to poll data");
+ }
+
+ }
+
+
+ @SuppressWarnings({"unchecked", "this-escape"})
+ private Producer<byte[], byte[]> expireProducer(ClusterInstance cluster) {
+ Map<String, Object> config = Map.of(
+ ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName(),
+ ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName(),
+ ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers(),
+ ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false,
+ ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 2000,
+ ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 1500
+ );
+ return new EvilKafkaProducerBuilder().build(config);
+ }
+
+ static class EvilKafkaProducerBuilder {
+
+ Serializer<byte[]> serializer = new ByteArraySerializer();
+ ApiVersions apiVersions = new ApiVersions();
+ LogContext logContext = new LogContext("[expire Producer test ]");
+ Metrics metrics = new Metrics(Time.SYSTEM);
+
+ String clientId;
+ String transactionalId;
+ ProducerConfig config;
+ ProducerMetadata metadata;
+ RecordAccumulator accumulator;
+ Partitioner partitioner;
+ Sender sender;
+ ProducerInterceptors<String, String> interceptors;
+
+ @SuppressWarnings({"unchecked", "this-escape"})
+ Producer<byte[], byte[]> build(Map<String, Object> configs) {
+ this.config = new ProducerConfig(ProducerConfig.appendSerializerToConfig(configs, null, null));
+ transactionalId = config.getString(ProducerConfig.TRANSACTIONAL_ID_CONFIG);
+ clientId = config.getString(ProducerConfig.CLIENT_ID_CONFIG);
+ return new KafkaProducer<>(
+ config,
+ logContext,
+ metrics,
+ serializer,
+ serializer,
+ buildMetadata(),
+ buildAccumulator(),
+ null,
+ buildSender(),
+ buildInterceptors(),
+ buildPartition(),
+ Time.SYSTEM,
+ ioThread(),
+ Optional.empty()
+ );
+ }
+
+
+ private ProducerInterceptors buildInterceptors() {
+ this.interceptors = new ProducerInterceptors<>(List.of(), metrics);
+ return this.interceptors;
+ }
+
+ private Partitioner buildPartition() {
+ this.partitioner = config.getConfiguredInstance(
+ ProducerConfig.PARTITIONER_CLASS_CONFIG,
+ Partitioner.class,
+ Collections.singletonMap(ProducerConfig.CLIENT_ID_CONFIG, clientId));
+ return this.partitioner;
+ }
+
+ private Sender buildSender() {
+ int maxInflightRequests = config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION);
+ int requestTimeoutMs = config.getInt(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG);
+ ProducerMetrics metricsRegistry = new ProducerMetrics(this.metrics);
+ Sensor throttleTimeSensor = Sender.throttleTimeSensor(metricsRegistry.senderMetrics);
+ KafkaClient client = ClientUtils.createNetworkClient(config,
+ this.metrics,
+ "producer",
+ logContext,
+ apiVersions,
+ Time.SYSTEM,
+ maxInflightRequests,
+ metadata,
+ throttleTimeSensor,
+ null);
+
+ short acks = Short.parseShort(config.getString(ProducerConfig.ACKS_CONFIG));
+ this.sender = new Sender(logContext,
+ client,
+ metadata,
+ this.accumulator,
+ maxInflightRequests == 1,
+ config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),
+ acks,
+ config.getInt(ProducerConfig.RETRIES_CONFIG),
+ metricsRegistry.senderMetrics,
+ Time.SYSTEM,
+ requestTimeoutMs,
+ config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG),
+ null) {
+ @Override
+ protected long sendProducerData(long now) {
+ long result = super.sendProducerData(now);
+ try {
+ // Ensure the batch expires.
+ Thread.sleep(500);
Review Comment:
Hmm, the delivery time is 2000ms. Is this enough to cause the batch to expire? Also, the integration test doesn't control when the inflight request is actually sent in the Socket layer. So, the test may not be very effective since the inflight request is likely sent to the Socket before the delivery time expires.
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerIntegrationTest.java:
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.producer;
+
+import org.apache.kafka.clients.ApiVersions;
+import org.apache.kafka.clients.ClientUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.KafkaClient;
+import org.apache.kafka.clients.producer.internals.BufferPool;
+import org.apache.kafka.clients.producer.internals.ProducerInterceptors;
+import org.apache.kafka.clients.producer.internals.ProducerMetadata;
+import org.apache.kafka.clients.producer.internals.ProducerMetrics;
+import org.apache.kafka.clients.producer.internals.RecordAccumulator;
+import org.apache.kafka.clients.producer.internals.Sender;
+import org.apache.kafka.common.compress.NoCompression;
+import org.apache.kafka.common.internals.ClusterResourceListeners;
+import org.apache.kafka.common.internals.Plugin;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.MetricsReporter;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.TestUtils;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
+
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+
+
+public class ProducerIntegrationTest {
+
+ @ClusterTest(serverProperties = {
+ @ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1"),
+ })
+ public void testInFlightBatchShouldNotBeCorrupted(ClusterInstance cluster) throws InterruptedException,
+ ExecutionException {
+ String topic = "test-topic";
+ cluster.createTopic("test-topic", 1, (short) 1);
+ try (var producer = expireProducer(cluster)) {
+ producer.send(new ProducerRecord<>(topic, "key".getBytes(), "value".getBytes())).get();
+ }
+ try (var consumer = cluster.consumer()) {
+ consumer.subscribe(List.of(topic));
+ TestUtils.waitForCondition(() -> consumer.poll(Duration.ofSeconds(1)).count() == 1, 5000, "failed to poll data");
+ }
+
Review Comment:
extra new line
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2644024076
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
In the current implementation, when failing a batch, we always (1) reset some states and fail the producer callback; (2) reclaim the buffer in the batch for reuse. This is causing an issue for failing the inflight batch when it hits the delivery time since we can't claim the buffer in the batch at that point. There are two potential ways to fix this issue.
The first one is what this PR does. It doesn't fail the inflight batch when it hits delivery time. instead, it waits for the inflight batch to hit the request timeout and then fail the batch (at this time, it's safe to reclaim the buffer in the batch). The benefit of this approach is that the fix is simple. The side effect is that in some rare cases, a record could take deliver time + request timeout to expire.
Another way is to continue to fail the inflight batch when it hits the delivery time. However, we only do item (1) above when failing the batch. When the inflight request times out, we do item (2) to reclaim the buffer. This keeps the delivery time guarantee, but might be a more complex fix.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3668022680
@junrao @chia7712
This PR affects two existing tests, testDropCommitOnBatchExpiry and testDuplicateSequenceAfterProducerReset in TransactionManagerTest. Both of them rely on in-flight batches, and Iâm not sure how to construct an expired batch under the new behavior. Could you help advise how these tests should be adapted?
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2634347713
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2686,14 +2687,12 @@ public void testResetNextBatchExpiry() throws Exception {
inOrder.verify(client, atLeastOnce()).newClientRequest(anyString(), any(), anyLong(), anyBoolean(), anyInt(), any());
inOrder.verify(client, atLeastOnce()).send(any(), anyLong());
inOrder.verify(client).poll(eq(0L), anyLong());
- inOrder.verify(client).poll(eq(accumulator.getDeliveryTimeoutMs()), anyLong());
Review Comment:
This part has been revised. Do you have an suggestions for expire batch, I'm troubled here @chia7712
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3674262348
@gongxuanzhang `testDropCommitOnBatchExpiry` should rely on the `NetworkClient`'s timeout handler rather than the `Sender`. Therefore, we should call `time.sleep(10000) ` after `sender.runOnce()` to ensure the `NetworkClient` expires the request. Note that the request is created inside `sender.runOnce()`, so its creation time is distinct from the batch's creation time
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2612457161
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:
##########
@@ -1602,13 +1602,34 @@ private MemoryRecords trimInvalidBytes(MemoryRecords records, LogAppendInfo info
if (validBytes == records.sizeInBytes()) {
return records;
} else {
- // trim invalid bytes
+ // Duplicate the original buffer for trimming and logging purposes.
ByteBuffer validByteBuffer = records.buffer().duplicate();
+
+ // Log detailed information about trimmed bytes
+ validByteBuffer.position(validBytes);
+ byte[] invalidBytes = new byte[records.sizeInBytes() - validBytes];
+ validByteBuffer.get(invalidBytes);
+ String invalidBytesHex = bytesToHex(invalidBytes);
+
+ logger.warn("Trimming invalid bytes from message set for partition {}. Original size: {} bytes, valid bytes: {}, trimmed bytes: {}. " +
Review Comment:
Yes, but we could probably just pass in a isLeader boolean.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3525136073
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3663956773
@chia7712
This PR affects two existing tests, `testDropCommitOnBatchExpiry` and `testDuplicateSequenceAfterProducerReset` in `TransactionManagerTest`. Both of them rely on in-flight batches, and Iâm not sure how to construct an expired batch under the new behavior. Could you help advise how these tests should be adapted?
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2644683412
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
The main concern with option 2 is the potential state inconsistency
For example, a user might receive a failure (timeout) via the callback, but the in-flight request actually succeeds. This mismatch between clientâs view and the serverâs state creates ambiguity and could lead to duplicate data if the user chooses to retry
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2645204209
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
I see. Thank youďź
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2644146066
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
For reference, https://github.com/apache/kafka/pull/21065 is an example of the second approach. Perhaps we could also take a look at that one and see if it's safe.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
For reference, https://github.com/apache/kafka/pull/21065 is an example of the second approach. Perhaps we could also take a look at that one and see if it's better.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3190508459
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3645844428
Not finished yet, please do not review.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3677059635
> It seems the CI has passed, but I'm not sure I understand the situation fully yet. Did we figure something out?
@jolshan : The main change in this PR is to not fail the inflight batches (in NetworkClient) on delivery time expiration. Instead, it waits until the request timeout and then fails the batch. Could you double check if the changes in TransactionManagerTest look ok?
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2625402300
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2642,11 +2645,11 @@ public void testExpiredBatchDoesNotRetry() throws Exception {
}
@Test
- public void testExpiredBatchDoesNotSplitOnMessageTooLargeError() throws Exception {
+ public void testBatchDoesNotSplitOnMessageTooLargeError() throws Exception {
Review Comment:
Now we no longer expire in-flight request batches, so Iâm not sure how to create an expired batch in order to test whether it gets split.
Can I just crudely sleep for the delivery timeout?
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2629420120
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2686,14 +2687,12 @@ public void testResetNextBatchExpiry() throws Exception {
inOrder.verify(client, atLeastOnce()).newClientRequest(anyString(), any(), anyLong(), anyBoolean(), anyInt(), any());
inOrder.verify(client, atLeastOnce()).send(any(), anyLong());
inOrder.verify(client).poll(eq(0L), anyLong());
- inOrder.verify(client).poll(eq(accumulator.getDeliveryTimeoutMs()), anyLong());
Review Comment:
During the process, poll is invoked with (maxLong - now), so I think the test should use gte(maxLong - System.currentTimeMillis()) for verification.
PTAL
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2593484432
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -354,7 +317,8 @@ private boolean shouldHandleAuthorizationError(RuntimeException exception) {
return false;
}
- private long sendProducerData(long now) {
+ // Visible for testing
+ protected long sendProducerData(long now) {
Review Comment:
The bug could also manifest as a different record batch being sent to the broker. In UnifiedLog.trimInvalidBytes(), we trim extra bytes from the record batch. On the leader side, we don't expect the record batch to have extra bytes. It would be useful to log a warning in this is not the case.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java:
##########
@@ -48,7 +48,7 @@ public class BufferPool {
private final long totalMemory;
private final int poolableSize;
- private final ReentrantLock lock;
+ protected final ReentrantLock lock;
Review Comment:
Is this change needed?
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2533,10 +2534,7 @@ public void testInflightBatchesExpireOnDeliveryTimeout() throws InterruptedExcep
time.sleep(deliveryTimeoutMs);
sender.runOnce(); // receive first response
assertEquals(0, sender.inFlightBatches(tp0).size(), "Expect zero in-flight batch in accumulator");
- assertInstanceOf(
- TimeoutException.class,
- assertThrows(ExecutionException.class, request::get).getCause(),
- "The expired batch should throw a TimeoutException");
+ assertDoesNotThrow(() -> request.get());
Review Comment:
Instead of a new integration test, I am wondering if we could just extend this test for coverage. We can create an inflight request, but not complete the request. Let the delivery timeout to expire and check that the batch is not completed.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -910,8 +874,11 @@ private void sendProduceRequest(long now, int destination, short acks, int timeo
RequestCompletionHandler callback = response -> handleProduceResponse(response, recordsByPartition, topicNames, time.milliseconds());
String nodeId = Integer.toString(destination);
+
+ long deliveryTimeoutMs = accumulator.getDeliveryTimeoutMs() - (now - earliestCreatedMs);
Review Comment:
Hmm, this is a bit more complicated than I thought since there are multiple batches with different creation time in a request. For now, it's probably better to just use requestTimeoutMs to make the fix simple. This could extend the delivery timeout by request timeout in the rare cases. We can file a followup jira to see if there are better solutions in the future.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -403,9 +367,7 @@ private long sendProducerData(long now) {
}
accumulator.resetNextBatchExpiryTime();
- List<ProducerBatch> expiredInflightBatches = getExpiredInflightBatches(now);
List<ProducerBatch> expiredBatches = this.accumulator.expiredBatches(now);
- expiredBatches.addAll(expiredInflightBatches);
Review Comment:
It would be useful to add a comment on why we are not expiring inflight batches 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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2644683413
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
The main concern with option 2 is the potential state inconsistency
For example, a user might receive a failure (timeout) via the callback, but the in-flight request actually succeeds. This mismatch between clientâs view and the serverâs state creates ambiguity and could lead to duplicate data if the user chooses to retry
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
The main concern with option 2 is the potential state inconsistency
For example, a user might receive a failure (timeout) via the callback, but the in-flight request actually succeeds. This mismatch between clientâs view and the serverâs state creates ambiguity and could lead to duplicate data if the user chooses to retry
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
The main concern with option 2 is the potential state inconsistency
For example, a user might receive a failure (timeout) via the callback, but the in-flight request actually succeeds. This mismatch between clientâs view and the serverâs state creates ambiguity and could lead to duplicate data if the user chooses to retry
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
The main concern with option 2 is the potential state inconsistency
For example, a user might receive a failure (timeout) via the callback, but the in-flight request actually succeeds. This mismatch between clientâs view and the serverâs state creates ambiguity and could lead to duplicate data if the user chooses to retry
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3732921356
closing this in favor of the approach in #21065, which delays returning the buffer
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3635623519
@junrao
Iâve updated the code to trim invalid bytes and log their content. One question I have: if the trimmed byte array is very large, should we consider limiting the length of the logged data? Would it make sense to introduce a configuration option for this to avoid excessively long logs
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3625153590
@junrao
https://github.com/apache/kafka/pull/19489/files#diff-51f6f7d29fde42e1e2ed2ade53b007201a34bcf9efd14f0e6e3fe8fe5114d892R59
Do you think this test is okay? I feel like I donât have a great way to test it.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2629403645
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2686,14 +2687,12 @@ public void testResetNextBatchExpiry() throws Exception {
inOrder.verify(client, atLeastOnce()).newClientRequest(anyString(), any(), anyLong(), anyBoolean(), anyInt(), any());
inOrder.verify(client, atLeastOnce()).send(any(), anyLong());
inOrder.verify(client).poll(eq(0L), anyLong());
- inOrder.verify(client).poll(eq(accumulator.getDeliveryTimeoutMs()), anyLong());
Review Comment:
@junrao
During the process, poll is invoked with (maxLong - now), so I think the test should use gte(maxLong - now) for verification.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3630322407
> https://github.com/apache/kafka/pull/19489/files#diff-51f6f7d29fde42e1e2ed2ade53b007201a34bcf9efd14f0e6e3fe8fe5114d892R59
>
> Do you think this test is okay? I feel like I donât have a great way to test it.
How about https://github.com/apache/kafka/pull/19489#discussion_r2593505677 ?
Also, could you address https://github.com/apache/kafka/pull/19489#discussion_r2593484432?
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3677493897
@gongxuanzhang would you mind creating a patch against the 3.9 branch as well?
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2637739892
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
We need to clarify in the docs that `delivery.timeout.ms` is no longer a hard limit if a request is already in-flight
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2629403645
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2686,14 +2687,12 @@ public void testResetNextBatchExpiry() throws Exception {
inOrder.verify(client, atLeastOnce()).newClientRequest(anyString(), any(), anyLong(), anyBoolean(), anyInt(), any());
inOrder.verify(client, atLeastOnce()).send(any(), anyLong());
inOrder.verify(client).poll(eq(0L), anyLong());
- inOrder.verify(client).poll(eq(accumulator.getDeliveryTimeoutMs()), anyLong());
Review Comment:
@junrao
During the process, poll is invoked with (maxLong - now), so I think the test should use gte(maxLong - now) for verification.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3675246277
@chia7712 @junrao @kirktrue
Thanks a lot for your review. The PR has now passed CI. PTAL
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2635905633
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -783,16 +785,18 @@ MAX_REQUEST_SIZE, ACKS_ALL, MAX_RETRIES, new SenderMetricsRegistry(metrics), thi
assertTrue(transactionManager.hasInflightBatches(tp0));
assertEquals(1, transactionManager.sequenceNumber(tp0));
- time.sleep(5000); // delivery time out
- sender.runOnce();
+ ProducerTestUtils.runUntil(sender, () -> {
+ time.sleep(5000); // advance the "now" so NetworkClient sees the request as expired
Review Comment:
> advance the "now" so NetworkClient sees the request as expired
advance the time so that NetworkClient will time out the request and cause the future to expire.
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2677,23 +2692,24 @@ public void testResetNextBatchExpiry() throws Exception {
appendToAccumulator(tp0, 0L, "key", "value");
sender.runOnce();
+ long nowBeforeSecondRun = time.milliseconds();
sender.runOnce();
- time.setCurrentTimeMs(time.milliseconds() + accumulator.getDeliveryTimeoutMs() + 1);
+ this.time.setCurrentTimeMs(this.time.milliseconds() + accumulator.getDeliveryTimeoutMs() + 1);
Review Comment:
Let's be consistent with referencing `time`. We don't use `this` two lines above.
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -783,16 +785,18 @@ MAX_REQUEST_SIZE, ACKS_ALL, MAX_RETRIES, new SenderMetricsRegistry(metrics), thi
assertTrue(transactionManager.hasInflightBatches(tp0));
assertEquals(1, transactionManager.sequenceNumber(tp0));
- time.sleep(5000); // delivery time out
- sender.runOnce();
+ ProducerTestUtils.runUntil(sender, () -> {
+ time.sleep(5000); // advance the "now" so NetworkClient sees the request as expired
+ return responseFuture1.isDone();
+ });
// The retried request will remain inflight until the request timeout
// is reached even though the delivery timeout has expired and the
// future has completed exceptionally.
assertTrue(responseFuture1.isDone());
- TestUtils.assertFutureThrowsWithMessageContaining(TimeoutException.class, responseFuture1, SENDER_TIMEOUT_MSG);
+ TestUtils.assertFutureThrowsWithMessageContaining(NetworkException.class, responseFuture1, NETWORK_DISCONNECTED_MSG);
Review Comment:
> // The retried request will remain inflight until the request timeout
> // is reached even though the delivery timeout has expired and the
> // future has completed exceptionally.
> assertTrue(responseFuture1.isDone());
These are no longer needed.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2638942073
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
Thank you for your reminder. I will make the necessary revisions
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2607602014
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2655,17 +2653,17 @@ public void testExpiredBatchDoesNotSplitOnMessageTooLargeError() throws Exceptio
client.respond(produceResponse(tp0, -1, Errors.MESSAGE_TOO_LARGE, -1));
time.sleep(deliverTimeoutMs);
- // expire the batch and process the response
+ // Does not verify expired batch
sender.runOnce();
- assertTrue(request1.isDone());
- assertTrue(request2.isDone());
assertEquals(0, client.inFlightRequestCount());
assertEquals(0, sender.inFlightBatches(tp0).size());
// run again and must not split big batch and resend anything.
sender.runOnce();
assertEquals(0, client.inFlightRequestCount());
assertEquals(0, sender.inFlightBatches(tp0).size());
+ assertTrue(request1.isDone());
Review Comment:
> // run again and must not split big batch and resend anything.
The code doesn't match the comment above. Also, the test fails now and needs to be adjusted.
```
org.opentest4j.AssertionFailedError:
Expected :0
Actual :1
<Click to see difference>
at org.junit.jupiter.api.AssertionFailureBuilder.build(AssertionFailureBuilder.java:151)
at org.junit.jupiter.api.AssertionFailureBuilder.buildAndThrow(AssertionFailureBuilder.java:132)
at org.junit.jupiter.api.AssertEquals.failNotEqual(AssertEquals.java:197)
at org.junit.jupiter.api.AssertEquals.assertEquals(AssertEquals.java:150)
at org.junit.jupiter.api.AssertEquals.assertEquals(AssertEquals.java:145)
at org.junit.jupiter.api.Assertions.assertEquals(Assertions.java:531)
at org.apache.kafka.clients.producer.internals.SenderTest.testExpiredBatchDoesNotSplitOnMessageTooLargeError(SenderTest.java:2663)
at java.base/java.lang.reflect.Method.invoke(Method.java:580)
```
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2714,12 +2712,8 @@ public void testExpiredBatchesInMultiplePartitions() throws Exception {
time.sleep(deliveryTimeoutMs);
sender.runOnce();
assertEquals(0, sender.inFlightBatches(tp0).size(), "Expect zero in-flight batch in accumulator");
-
- ExecutionException e = assertThrows(ExecutionException.class, request1::get);
- assertInstanceOf(TimeoutException.class, e.getCause());
-
- e = assertThrows(ExecutionException.class, request2::get);
- assertInstanceOf(TimeoutException.class, e.getCause());
+ assertDoesNotThrow(() -> request1.get());
Review Comment:
Same comment for `testInflightBatchesExpireOnDeliveryTimeout`.
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2533,10 +2534,7 @@ public void testInflightBatchesExpireOnDeliveryTimeout() throws InterruptedExcep
time.sleep(deliveryTimeoutMs);
sender.runOnce(); // receive first response
assertEquals(0, sender.inFlightBatches(tp0).size(), "Expect zero in-flight batch in accumulator");
- assertInstanceOf(
- TimeoutException.class,
- assertThrows(ExecutionException.class, request::get).getCause(),
- "The expired batch should throw a TimeoutException");
+ assertDoesNotThrow(() -> request.get());
Review Comment:
1. We need to change the test name since now we don't expire inflight requests.
2. We can modify this test to provide better coverage. We can avoid sending the client response first. Let the expiration time expire and check that the inflight request remains uncompleted. We then send the client response and check that the inflight request is completed. With this test, we don't need ProducerIntegrationTest.
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:
##########
@@ -1602,13 +1602,34 @@ private MemoryRecords trimInvalidBytes(MemoryRecords records, LogAppendInfo info
if (validBytes == records.sizeInBytes()) {
return records;
} else {
- // trim invalid bytes
+ // Duplicate the original buffer for trimming and logging purposes.
ByteBuffer validByteBuffer = records.buffer().duplicate();
+
+ // Log detailed information about trimmed bytes
+ validByteBuffer.position(validBytes);
+ byte[] invalidBytes = new byte[records.sizeInBytes() - validBytes];
+ validByteBuffer.get(invalidBytes);
+ String invalidBytesHex = bytesToHex(invalidBytes);
+
+ logger.warn("Trimming invalid bytes from message set for partition {}. Original size: {} bytes, valid bytes: {}, trimmed bytes: {}. " +
Review Comment:
1. We only need to log the amount of trimmed bytes. We don't need to log the content since it's typically binary and not meaningful.
2. Appends from the follower could have trimmed bytes. So we only want to log the warning if the append comes from the leader.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2642191332
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
Expiring an in-flight batch causes the buffer to be reused,leading to data corruption in the active request.
Given that the request-level timeout already handles request expiration, removing the hard limit for batch-level timeout appears to be a reasonable approach
The side effect is that a batch can live longer than the configured delivery timeout, since the batch is effectively exempt from the batch-level check while in-flight
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2644118002
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
I would vote for the second option. Changing the meaning of delivery timeout has very broad impact and would not be appropriate for a patch release.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2634356916
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2686,14 +2687,12 @@ public void testResetNextBatchExpiry() throws Exception {
inOrder.verify(client, atLeastOnce()).newClientRequest(anyString(), any(), anyLong(), anyBoolean(), anyInt(), any());
inOrder.verify(client, atLeastOnce()).send(any(), anyLong());
inOrder.verify(client).poll(eq(0L), anyLong());
- inOrder.verify(client).poll(eq(accumulator.getDeliveryTimeoutMs()), anyLong());
Review Comment:
see https://github.com/apache/kafka/pull/19489#issuecomment-3674262348
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2625550884
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2642,11 +2645,11 @@ public void testExpiredBatchDoesNotRetry() throws Exception {
}
@Test
- public void testExpiredBatchDoesNotSplitOnMessageTooLargeError() throws Exception {
+ public void testBatchDoesNotSplitOnMessageTooLargeError() throws Exception {
Review Comment:
The split batches are created automatically in Sender. You can probably just follow the following code in testTooLargeBatchesAreSafelyRemoved() to send extra produce responses after the split.
```
// send request
sender.runOnce();
assertEquals(1, sender.inFlightBatches(tp0).size());
// return a MESSAGE_TOO_LARGE error
client.respond(produceResponse(tp0, -1, Errors.MESSAGE_TOO_LARGE, -1));
sender.runOnce();
// process retried response
sender.runOnce();
client.respond(produceResponse(tp0, 0, Errors.NONE, 0));
sender.runOnce();
// In-flight batches should be empty. Sleep past the expiration time of the batch and run once, no error should be thrown
assertEquals(0, sender.inFlightBatches(tp0).size());
```
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3630005543
@gongxuanzhang could you merge trunk to trigger the CI again?
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2641260972
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -783,16 +785,18 @@ MAX_REQUEST_SIZE, ACKS_ALL, MAX_RETRIES, new SenderMetricsRegistry(metrics), thi
assertTrue(transactionManager.hasInflightBatches(tp0));
assertEquals(1, transactionManager.sequenceNumber(tp0));
- time.sleep(5000); // delivery time out
- sender.runOnce();
+ ProducerTestUtils.runUntil(sender, () -> {
+ time.sleep(5000); // advance the "now" so NetworkClient sees the request as expired
+ return responseFuture1.isDone();
+ });
// The retried request will remain inflight until the request timeout
// is reached even though the delivery timeout has expired and the
// future has completed exceptionally.
assertTrue(responseFuture1.isDone());
- TestUtils.assertFutureThrowsWithMessageContaining(TimeoutException.class, responseFuture1, SENDER_TIMEOUT_MSG);
+ TestUtils.assertFutureThrowsWithMessageContaining(NetworkException.class, responseFuture1, NETWORK_DISCONNECTED_MSG);
assertFalse(transactionManager.hasInFlightRequest());
- assertEquals(1, client.inFlightRequestCount());
+ assertEquals(0, client.inFlightRequestCount());
Review Comment:
I think for me, the main thing I'm not sure about is if this test is even testing the original behavior. It is a bit hard to parse what the original behavior was though. It says `testDuplicateSequenceAfterProducerReset` but I'm not even sure where this duplicate sequence is supposed to happen.
The request remaining inflight also seems like it would matter, but now that is no longer the case.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3610542000
> Have you seen [KAFKA-19012](https://issues.apache.org/jira/browse/KAFKA-19012)? I'm wondering if there's any relationship between it and [KAFKA-17862](https://issues.apache.org/jira/browse/KAFKA-17862) đ¤
Yes, they encountered the same bug.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2579223228
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
- List<ProducerBatch> expiredBatches = new ArrayList<>();
-
- for (Iterator<Map.Entry<TopicPartition, List<ProducerBatch>>> batchIt = inFlightBatches.entrySet().iterator(); batchIt.hasNext();) {
- Map.Entry<TopicPartition, List<ProducerBatch>> entry = batchIt.next();
- List<ProducerBatch> partitionInFlightBatches = entry.getValue();
- if (partitionInFlightBatches != null) {
- Iterator<ProducerBatch> iter = partitionInFlightBatches.iterator();
- while (iter.hasNext()) {
- ProducerBatch batch = iter.next();
- if (batch.hasReachedDeliveryTimeout(accumulator.getDeliveryTimeoutMs(), now)) {
- iter.remove();
- // expireBatches is called in Sender.sendProducerData, before client.poll.
- // The !batch.isDone() invariant should always hold. An IllegalStateException
- // exception will be thrown if the invariant is violated.
- if (!batch.isDone()) {
- expiredBatches.add(batch);
- } else {
- throw new IllegalStateException(batch.topicPartition + " batch created at " +
- batch.createdMs + " gets unexpected final state " + batch.finalState());
- }
- } else {
- accumulator.maybeUpdateNextBatchExpiryTime(batch);
- break;
- }
- }
- if (partitionInFlightBatches.isEmpty()) {
- batchIt.remove();
- }
- }
- }
- return expiredBatches;
- }
-
Review Comment:
It would be good to get this running with the continuous integration to ensure it doesn't cause problems elsewhere.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2644748236
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
I understand that network uncertainty is unavoidable, as you mentioned. However, there is a key distinction between unavoidable network ambiguity and intentional state divergence. We are actively creating a scenario where the client logic 'gives up' while the (kafka) I/O layer continues to process the request
That said, it appears intentional divergence was indeed anticipated by the KIP-91.
> This implies that user might be notified of batch expiry while a batch is still in-flight.
While I fail to see the practical benefit of this use case, I am amenable to preserving this behavior since it has been documented and released for a long time.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-2826210623
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-2837333081
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 closed pull request #19489: KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool
URL: https://github.com/apache/kafka/pull/19489
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2645170449
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
Perhaps I can modify the bufferpool in this PR?
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2644024076
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
In the current implementation, when failing a batch, we always (1) reset some states and fail the producer callback; (2) reclaim the buffer in the batch for reuse. This is causing an issue for failing the inflight batch when it hits the delivery time since we can't claim the buffer in the batch at that point. There are two potential ways to fix this issue.
The first one is what this PR does. It doesn't fail the inflight batch when it hits delivery time. instead, it waits for the inflight batch to hit the request timeout and then fail the batch (at this time, it's safe to reclaim the buffer in the batch). The benefit of this approach is that the fix is simple. The side effect is that in some rare cases, a record could take deliver time + request timeout to expire.
Another way is to continue to fail the inflight batch when it hits the delivery time. However, we only do item (1) above when failing the batch. When the inflight request times out, we do item (2) to reclaim the buffer. This keeps the delivery time guarantee, but is a more complex fix.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2645182103
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -174,43 +174,6 @@ private void maybeRemoveAndDeallocateBatch(ProducerBatch batch) {
this.accumulator.deallocate(batch);
}
- /**
- * Get the in-flight batches that has reached delivery timeout.
- */
- private List<ProducerBatch> getExpiredInflightBatches(long now) {
Review Comment:
I just left a comment on https://github.com/apache/kafka/pull/21065#discussion_r2645158450
> Perhaps I can modify the bufferpool in this PR?
I prefer letting this PR implement the option 1, and we can use #21065 to explore the another 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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2624340885
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2526,17 +2526,20 @@ public void testInflightBatchesExpireOnDeliveryTimeout() throws InterruptedExcep
assertEquals(1, client.inFlightRequestCount());
assertEquals(1, sender.inFlightBatches(tp0).size(), "Expect one in-flight batch in accumulator");
+ // let delivery timeout expire without responding
+ time.sleep(deliveryTimeoutMs);
+ sender.runOnce();
+ // inflight batch should still be there
+ assertEquals(1, sender.inFlightBatches(tp0).size());
+ assertFalse(request.isDone());
+
Map<TopicIdPartition, ProduceResponse.PartitionResponse> responseMap = new HashMap<>();
responseMap.put(new TopicIdPartition(TOPIC_ID, tp0), new ProduceResponse.PartitionResponse(Errors.NONE, 0L, 0L, 0L));
client.respond(new ProduceResponse(responseMap));
- time.sleep(deliveryTimeoutMs);
sender.runOnce(); // receive first response
- assertEquals(0, sender.inFlightBatches(tp0).size(), "Expect zero in-flight batch in accumulator");
- assertInstanceOf(
- TimeoutException.class,
- assertThrows(ExecutionException.class, request::get).getCause(),
- "The expired batch should throw a TimeoutException");
+ assertEquals(0, sender.inFlightBatches(tp0).size());
Review Comment:
This is an existing problem. Should we remove `first` in the line above since there is only 1 response.
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2642,11 +2645,11 @@ public void testExpiredBatchDoesNotRetry() throws Exception {
}
@Test
- public void testExpiredBatchDoesNotSplitOnMessageTooLargeError() throws Exception {
+ public void testBatchDoesNotSplitOnMessageTooLargeError() throws Exception {
Review Comment:
This name is inaccurate since the batch does split in this case. How about testExpiredBatchStillSplitOnMessageTooLargeError ?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:
##########
@@ -1593,7 +1596,7 @@ private boolean hasHigherPartitionLeaderEpoch(RecordBatch batch, AppendOrigin or
* @param info The general information of the message set
* @return A trimmed message set. This may be the same as what was passed in, or it may not.
*/
- private MemoryRecords trimInvalidBytes(MemoryRecords records, LogAppendInfo info) {
+ private MemoryRecords trimInvalidBytes(MemoryRecords records, LogAppendInfo info, boolean isLeader) {
Review Comment:
Could we add the new param to javadoc?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:
##########
@@ -1605,6 +1608,10 @@ private MemoryRecords trimInvalidBytes(MemoryRecords records, LogAppendInfo info
// trim invalid bytes
ByteBuffer validByteBuffer = records.buffer().duplicate();
validByteBuffer.limit(validBytes);
+ if (isLeader) {
+ logger.warn("Trimming invalid bytes from message set for partition {}. Original size: {} bytes, valid bytes: {}, trimmed bytes: {}. ",
Review Comment:
We have the following log ident. So, there is not need to include partition explicitly.
`this.logIdent = "[UnifiedLog partition=" + topicPartition() + ", dir=" + parentDir() + "] ";`
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2655,17 +2658,15 @@ public void testExpiredBatchDoesNotSplitOnMessageTooLargeError() throws Exceptio
client.respond(produceResponse(tp0, -1, Errors.MESSAGE_TOO_LARGE, -1));
time.sleep(deliverTimeoutMs);
- // expire the batch and process the response
sender.runOnce();
- assertTrue(request1.isDone());
- assertTrue(request2.isDone());
- assertEquals(0, client.inFlightRequestCount());
- assertEquals(0, sender.inFlightBatches(tp0).size());
+ sender.runOnce();
Review Comment:
It would be useful to document the behavior after each runOnce(). After the first one, we expect the original batch to be split. After the second one, we expect the first split batch to be sent.
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:
##########
@@ -1100,7 +1102,8 @@ private LogAppendInfo append(MemoryRecords records,
VerificationGuard verificationGuard,
boolean ignoreRecordSize,
byte toMagic,
- short transactionVersion) {
+ short transactionVersion,
+ boolean isLeader) {
Review Comment:
We don't need to pass in isLeader. isLeader can be computed as origin != REPLICATION.
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2686,14 +2687,12 @@ public void testResetNextBatchExpiry() throws Exception {
inOrder.verify(client, atLeastOnce()).newClientRequest(anyString(), any(), anyLong(), anyBoolean(), anyInt(), any());
inOrder.verify(client, atLeastOnce()).send(any(), anyLong());
inOrder.verify(client).poll(eq(0L), anyLong());
- inOrder.verify(client).poll(eq(accumulator.getDeliveryTimeoutMs()), anyLong());
Review Comment:
It seems that the next two poll() calls should use maxLong as the poll time. We should verify them. However, if I add the verification with maxLong, the test fails. Do you know why @gongxuanzhang and @kirktrue ?
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2642,11 +2645,11 @@ public void testExpiredBatchDoesNotRetry() throws Exception {
}
@Test
- public void testExpiredBatchDoesNotSplitOnMessageTooLargeError() throws Exception {
+ public void testBatchDoesNotSplitOnMessageTooLargeError() throws Exception {
long deliverTimeoutMs = 1500L;
// create a producer batch with more than one record so it is eligible for splitting
- Future<RecordMetadata> request1 = appendToAccumulator(tp0);
- Future<RecordMetadata> request2 = appendToAccumulator(tp0);
+ appendToAccumulator(tp0);
Review Comment:
It would be useful to track those futures, complete the responses after the split and verify that neither future throws.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on PR #19489:
URL: https://github.com/apache/kafka/pull/19489#issuecomment-3672786813
> This PR affects two existing tests, testDropCommitOnBatchExpiry and testDuplicateSequenceAfterProducerReset in TransactionManagerTest. Both of them rely on in-flight batches, and Iâm not sure how to construct an expired batch under the new behavior. Could you help advise how these tests should be adapted?
@artemlivshits @jolshan : Any recommendations on fixing the tests in TransactionManagerTest due to the changes in this PR?
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2628062038
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2642,30 +2645,41 @@ public void testExpiredBatchDoesNotRetry() throws Exception {
}
@Test
- public void testExpiredBatchDoesNotSplitOnMessageTooLargeError() throws Exception {
- long deliverTimeoutMs = 1500L;
+ public void testBatchStillSplitOnMessageTooLargeError() throws Exception {
Review Comment:
testBatchStillSplitOnMessageTooLargeError => testExpiredBatchStillSplitOnMessageTooLargeError
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2686,14 +2687,12 @@ public void testResetNextBatchExpiry() throws Exception {
inOrder.verify(client, atLeastOnce()).newClientRequest(anyString(), any(), anyLong(), anyBoolean(), anyInt(), any());
inOrder.verify(client, atLeastOnce()).send(any(), anyLong());
inOrder.verify(client).poll(eq(0L), anyLong());
- inOrder.verify(client).poll(eq(accumulator.getDeliveryTimeoutMs()), anyLong());
Review Comment:
This comment still needs to be addressed.
--
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
Re: [PR] KAFKA-17862: [buffer pool] corruption during buffer reuse from the pool [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19489:
URL: https://github.com/apache/kafka/pull/19489#discussion_r2634218117
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -2686,14 +2687,12 @@ public void testResetNextBatchExpiry() throws Exception {
inOrder.verify(client, atLeastOnce()).newClientRequest(anyString(), any(), anyLong(), anyBoolean(), anyInt(), any());
inOrder.verify(client, atLeastOnce()).send(any(), anyLong());
inOrder.verify(client).poll(eq(0L), anyLong());
- inOrder.verify(client).poll(eq(accumulator.getDeliveryTimeoutMs()), anyLong());
Review Comment:
> poll is invoked with (maxLong - now)
Yes, `nextExpiryTimeMs` is reset to `Long.MAX_VALUE`, and the timeout is calculated as `nextExpiryTimeMs` minus `now`
```java
pollTimeout = Math.min(pollTimeout, this.accumulator.nextExpiryTimeMs() - now);
```
@gongxuanzhang Is `inOrder.verify(client).poll(eq(accumulator.nextExpiryTimeMs() - nowBeforeSecondRun), anyLong());` more suitable?
--
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