You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/05/01 08:53:22 UTC
[GitHub] [pulsar] BewareMyPower opened a new pull request, #15406: [Java Client] Fix flaky BatchMessageTest by initializing lastBatchSendNanoTime
BewareMyPower opened a new pull request, #15406:
URL: https://github.com/apache/pulsar/pull/15406
### Motivation
`BatchMessageTest` is very flaky, some tests failed at checking the
backlog of a topic. This flakiness, maybe the bug, was introduced from
https://github.com/apache/pulsar/pull/14185. `batchFlushTask` calculates
the time duration between now and `lastBatchSendNanoTime`. However,
for the first time `batchFlushTask` is called, `lastBatchSendNanoTime`
is not initialized, so the time diff duration satified so that
`batchMessageAndSend` will always be called for the first time.
For example, if all messages can be split into two batches for a given
batch size limit, there is a chance that when first few messages are in
the batch, `batchFlushTask` is called at the same time and these
messages could be sent first.
Here are some extra logs added by myself:
```
2022-05-01T12:35:23,545+0800 [main] INFO org.apache.pulsar.client.impl.BatchMessageKeyBasedContainer - XYZ add 0 | 1 (0) 12 (600) | false
2022-05-01T12:35:23,545+0800 [pulsar-client-io-35-1] INFO org.apache.pulsar.client.impl.ProducerImpl - XYZ batchFlushTask 52779290583 us (5000000)
2022-05-01T12:35:23,545+0800 [pulsar-client-io-35-1] INFO org.apache.pulsar.client.impl.BatchMessageKeyBasedContainer - XYZ before create: [0]
2022-05-01T12:35:23,546+0800 [pulsar-client-io-35-1] INFO org.apache.pulsar.client.impl.BatchMessageKeyBasedContainer - XYZ create 1 Ops: 1 (0) 12 (600)
```
We can see `batchFlushTask` was called while the batch contains only 1
message. The time duration is nearly 15 hours, which is appearly wrong.
### Modifications
Initialize `lastBatchSendNanoTime` with the current time point
immediately after the batch container is constructed.
### Documentation
Check the box below or label this PR directly.
Need to update docs?
- [ ] `doc-required`
(Your PR needs to update docs and you will update later)
- [x] `no-need-doc`
(Please explain why)
- [ ] `doc`
(Your PR contains doc changes)
- [ ] `doc-added`
(Docs have been already added)
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on pull request #15406: [Java Client] Fix flaky BatchMessageTest by initializing lastBatchSendNanoTime
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on PR #15406:
URL: https://github.com/apache/pulsar/pull/15406#issuecomment-1114181869
Apply the following diff to master branch.
```diff
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
index afd3f808f8e..fb0c4cf9da4 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
@@ -23,6 +23,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.stream.Collectors;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.impl.ProducerImpl.OpSendMsg;
import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
@@ -102,6 +103,9 @@ class BatchMessageContainerImpl extends AbstractBatchMessageContainer {
highestSequenceId = msg.getSequenceId();
ProducerImpl.LAST_SEQ_ID_PUSHED_UPDATER.getAndUpdate(producer, prev -> Math.max(prev, msg.getSequenceId()));
+ log.info("XYZ add {} | {} ({}) {} ({}) | {}", msg.getSequenceId(),
+ numMessagesInBatch, maxNumMessagesInBatch,
+ currentBatchSizeBytes, maxBytesInBatch, isBatchFull());
return isBatchFull();
}
@@ -183,6 +187,8 @@ class BatchMessageContainerImpl extends AbstractBatchMessageContainer {
@Override
public OpSendMsg createOpSendMsg() throws IOException {
+ log.info("XYZ before create: {}",
+ messages.stream().map(MessageImpl::getSequenceId).collect(Collectors.toList()));
ByteBuf encryptedPayload = producer.encryptMessage(messageMetadata, getCompressedBatchMetadataAndPayload());
if (encryptedPayload.readableBytes() > ClientCnx.getMaxMessageSize()) {
discard(new PulsarClientException.InvalidMessageException(
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java
index 7614728353f..f4d7dc3afb0 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java
@@ -28,6 +28,7 @@ import java.util.Base64;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
import org.apache.pulsar.common.api.proto.CompressionType;
@@ -81,6 +82,9 @@ class BatchMessageKeyBasedContainer extends AbstractBatchMessageContainer {
} else {
part.addMsg(msg, callback);
}
+ log.info("XYZ add {} | {} ({}) {} ({}) | {}", msg.getSequenceId(),
+ numMessagesInBatch, maxNumMessagesInBatch,
+ currentBatchSizeBytes, maxBytesInBatch, isBatchFull());
return isBatchFull();
}
@@ -116,6 +120,8 @@ class BatchMessageKeyBasedContainer extends AbstractBatchMessageContainer {
}
private ProducerImpl.OpSendMsg createOpSendMsg(KeyedBatch keyedBatch) throws IOException {
+ log.info("XYZ before create: {}",
+ keyedBatch.messages.stream().map(MessageImpl::getSequenceId).collect(Collectors.toList()));
ByteBuf encryptedPayload = producer.encryptMessage(keyedBatch.messageMetadata,
keyedBatch.getCompressedBatchMetadataAndPayload());
if (encryptedPayload.readableBytes() > ClientCnx.getMaxMessageSize()) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index 7c3e7d1b0c1..57bac50c1e3 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -2054,6 +2054,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
// If a batch was sent more recently than the BatchingMaxPublishDelayMicros, schedule another flush to run just
// at BatchingMaxPublishDelayMicros after the last send.
long microsSinceLastSend = TimeUnit.NANOSECONDS.toMicros(System.nanoTime() - lastBatchSendNanoTime);
+ log.info("XYZ batchFlushTask {} us ({})", microsSinceLastSend, conf.getBatchingMaxPublishDelayMicros());
if (microsSinceLastSend < conf.getBatchingMaxPublishDelayMicros()) {
scheduleBatchFlushTask(conf.getBatchingMaxPublishDelayMicros() - microsSinceLastSend);
return;
@@ -2303,4 +2304,4 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
}
private static final Logger log = LoggerFactory.getLogger(ProducerImpl.class);
-}
\ No newline at end of file
+}
```
Run `BatchMessageTest#testSimpleBatchProducerWithFixedBatchBytes` for multiple times until it failed (adding the logs might reduce the flakiness), we can see following logs in a test.
```
2022-05-01T17:17:35,775+0800 [main] INFO org.apache.pulsar.client.impl.BatchMessageContainerImpl - XYZ add 0 | 1 (0) 12 (600) | false
2022-05-01T17:17:35,775+0800 [pulsar-client-io-35-1] INFO org.apache.pulsar.client.impl.ProducerImpl - XYZ batchFlushTask 61709385067 us (5000000)
2022-05-01T17:17:35,775+0800 [pulsar-client-io-35-1] INFO org.apache.pulsar.client.impl.BatchMessageContainerImpl - XYZ before create: [0]
2022-05-01T17:17:35,784+0800 [main] INFO org.apache.pulsar.client.impl.BatchMessageContainerImpl - XYZ add 1 | 1 (0) 12 (600) | false
2022-05-01T17:17:35,784+0800 [main] INFO org.apache.pulsar.client.impl.BatchMessageContainerImpl - XYZ add 2 | 2 (0) 24 (600) | false
...
2022-05-01T17:17:35,789+0800 [main] INFO org.apache.pulsar.client.impl.BatchMessageContainerImpl - XYZ add 46 | 46 (0) 589 (600) | false
2022-05-01T17:17:35,789+0800 [main] INFO org.apache.pulsar.client.impl.BatchMessageContainerImpl - XYZ before create: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46]
2022-05-01T17:17:35,793+0800 [main] INFO org.apache.pulsar.client.impl.BatchMessageContainerImpl - XYZ add 47 | 1 (0) 13 (600) | false
2022-05-01T17:17:35,794+0800 [main] INFO org.apache.pulsar.client.impl.BatchMessageContainerImpl - XYZ add 48 | 2 (0) 26 (600) | false
2022-05-01T17:17:35,794+0800 [main] INFO org.apache.pulsar.client.impl.BatchMessageContainerImpl - XYZ add 49 | 3 (0) 39 (600) | false
2022-05-01T17:17:39,089+0800 [pulsar-load-manager-3-1] INFO org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl - Writing local data to metadata store because maximum change 53.457931891452624% exceeded threshold 10%; time since last report written is 4.999 seconds
2022-05-01T17:17:39,093+0800 [metadata-store-11-1] INFO org.apache.pulsar.metadata.coordination.impl.ResourceLockImpl - Acquired resource lock on /loadbalance/brokers/localhost:61921
2022-05-01T17:17:40,784+0800 [pulsar-client-io-35-1] INFO org.apache.pulsar.client.impl.ProducerImpl - XYZ batchFlushTask 4995359 us (5000000)
2022-05-01T17:17:40,789+0800 [pulsar-client-io-35-1] INFO org.apache.pulsar.client.impl.ProducerImpl - XYZ batchFlushTask 5000690 us (5000000)
2022-05-01T17:17:40,790+0800 [pulsar-client-io-35-1] INFO org.apache.pulsar.client.impl.BatchMessageContainerImpl - XYZ before create: [47, 48, 49]
java.lang.AssertionError: expected [2] but found [3]
Expected :2
Actual :3
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on pull request #15406: [Java Client] Fix flaky BatchMessageTest by initializing lastBatchSendNanoTime
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on PR #15406:
URL: https://github.com/apache/pulsar/pull/15406#issuecomment-1114175051
To explain my extra logs, `1 (0) 12 (600)` means there is 1 message and 12 bytes in the batch, while the number limit and size limit are 0 and 600 bytes.
Before this patch, `BatchMessageTest` failed nearly each time in my local env.
<img width="1260" alt="image" src="https://user-images.githubusercontent.com/18204803/166138957-c73e9a57-e4b4-44a9-bd89-e15dc6565544.png">
I'm not sure whether this test class is ignored in CI so that CI could pass.
After this patch, I've run the test for 3 times in a row and it always succeeded.
<img width="1016" alt="image" src="https://user-images.githubusercontent.com/18204803/166139045-cf1b2947-63f2-4b3a-8282-271aa6dad881.png">
I found https://github.com/apache/pulsar/pull/14968 also tried to fix this flaky test, please also take a look, @nodece
A simpler fix solution is calling `producer.flush()` in these tests so that the batch won't wait until the batch timer task is scheduled. But it will hide the bug that `lastBatchSendNanoTime` was not initialized.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on pull request #15406: [Java Client] Fix flaky BatchMessageTest by initializing lastBatchSendNanoTime
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on PR #15406:
URL: https://github.com/apache/pulsar/pull/15406#issuecomment-1114268024
I removed the `release/2.10.1` label because this PR is a patch on #14185, which doesn't have the `release/2.10.1` label.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower merged pull request #15406: [Java Client] Fix flaky BatchMessageTest by initializing lastBatchSendNanoTime
Posted by GitBox <gi...@apache.org>.
BewareMyPower merged PR #15406:
URL: https://github.com/apache/pulsar/pull/15406
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on pull request #15406: [Java Client] Fix flaky BatchMessageTest by initializing lastBatchSendNanoTime
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on PR #15406:
URL: https://github.com/apache/pulsar/pull/15406#issuecomment-1211529705
@BewareMyPower - this is a really late review. Thank you for fixing my flaky test. I'd like to propose a slightly different implementation. Please take a look at https://github.com/apache/pulsar/pull/17058, if you're able. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org