You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "Demogorgon314 (via GitHub)" <gi...@apache.org> on 2023/08/30 05:42:11 UTC
[GitHub] [pulsar] Demogorgon314 opened a new pull request, #21091: [fix][broker] Avoid send batch message to loadbalancer service unit state topic to avoid split message to two batches
Demogorgon314 opened a new pull request, #21091:
URL: https://github.com/apache/pulsar/pull/21091
### Motivation
Currently, the `ServiceUnitStateChannelImpl` will send batch messages to `loadbalancer-service-unit-state` topic,
and the `StrategicTwoPhaseCompactor` will consume the message and rebatch the message to a new batch.
However, the `StrategicTwoPhaseCompactor` may write messages from the same batch into different entries, which will result in skipping some messages when reading compacted entries.
### Modifications
* Disable batching when writing message to `loadbalancer-service-unit-state` topic.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- 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] Demogorgon314 commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312634619
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -90,6 +91,23 @@ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
this.cryptoKeyReader = cryptoKeyReader;
}
+ @Override
+ public boolean add(MessageImpl<?> msg, SendCallback callback) {
+ this.lastAddedMessageId = (MessageIdImpl) msg.getMessageId();
+ return super.add(msg, callback);
+ }
+
+ @Override
+ public boolean haveEnoughSpace(MessageImpl<?> msg) {
+ if (lastAddedMessageId == null) {
Review Comment:
Yes, when no message is added, it will be null.
--
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 a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312783739
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -44,16 +44,17 @@
* [(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]
*/
public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
- MessageCrypto msgCrypto;
- Set<String> encryptionKeys;
- CryptoKeyReader cryptoKeyReader;
+ private MessageCrypto<MessageMetadata, MessageMetadata> msgCrypto;
+ private Set<String> encryptionKeys;
+ private CryptoKeyReader cryptoKeyReader;
+ private MessageIdImpl lastAddedMessageId;
- public RawBatchMessageContainerImpl(int maxNumMessagesInBatch, int maxBytesInBatch) {
+ public RawBatchMessageContainerImpl() {
super();
this.compressionType = CompressionType.NONE;
this.compressor = new CompressionCodecNone();
- this.maxNumMessagesInBatch = maxNumMessagesInBatch;
- this.maxBytesInBatch = maxBytesInBatch;
+ this.maxNumMessagesInBatch = Integer.MAX_VALUE;
+ this.maxBytesInBatch = Integer.MAX_VALUE;
Review Comment:
> Override is not working for the parent class.
I think it works and I just verified it with the following demo.
```java
class Base {
public boolean f() {
return g();
}
protected boolean g() {
return false;
}
}
class Derived extends Base {
@Override
public boolean f() {
return super.f();
}
@Override
public boolean g() {
return true;
}
}
public class Main {
public static void main(String[] args) {
System.out.println(new Base().f());
System.out.println(new Derived().f());
}
}
```
Output:
```
false
true
```
--
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 a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312612893
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -44,16 +44,17 @@
* [(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]
*/
public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
- MessageCrypto msgCrypto;
- Set<String> encryptionKeys;
- CryptoKeyReader cryptoKeyReader;
+ private MessageCrypto<MessageMetadata, MessageMetadata> msgCrypto;
+ private Set<String> encryptionKeys;
+ private CryptoKeyReader cryptoKeyReader;
+ private MessageIdImpl lastAddedMessageId;
Review Comment:
If this commit is not to be cherry-picked into branch-2.11 or earlier, use `MessageIdAdv` instead.
--
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] Demogorgon314 commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312836397
##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/StrategicCompactionTest.java:
##########
@@ -148,24 +160,65 @@ public void testNumericOrderCompaction() throws Exception {
Assert.assertEquals(tableView.entrySet(), expectedCopy.entrySet());
}
- @Override
- public void testCompactCompressedBatching() throws Exception {
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 10);
- super.testCompactCompressedBatching();
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 1);
- }
+ @Test(timeOut = 20000)
+ public void testSameBatchCompactToSameBatch() throws Exception {
+ final String topic =
+ "persistent://my-property/use/my-ns/testSameBatchCompactToSameBatch" + UUID.randomUUID();
- @Override
- public void testCompactEncryptedAndCompressedBatching() throws Exception {
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 10);
- super.testCompactEncryptedAndCompressedBatching();
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 1);
- }
+ final int messages = 10;
+
+ // 1.create producer and publish message to the topic.
+ ProducerBuilder<Integer> builder = pulsarClient.newProducer(Schema.INT32)
+ .compressionType(MSG_COMPRESSION_TYPE).topic(topic);
+ builder.batchingMaxMessages(2)
+ .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS);
+
+ Producer<Integer> producer = builder.create();
+
+ List<CompletableFuture<MessageId>> futures = new ArrayList<>(messages);
+ for (int i = 0; i < messages; i++) {
+ futures.add(producer.newMessage().key(String.valueOf(i))
+ .value(i)
+ .sendAsync());
+ }
+ // Add additional message to make sure the last batch is not full and test flush this batch.
+ futures.add(producer.newMessage().key(String.valueOf(messages))
+ .value(messages)
+ .sendAsync());
+
+ FutureUtil.waitForAll(futures).get();
+
+ // 2.compact the topic.
+ StrategicTwoPhaseCompactor compactor
+ = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+ compactor.compact(topic, strategy).get();
+
+ // consumer with readCompacted enabled only get compacted entries
+ try (Consumer<Integer> consumer = pulsarClient
+ .newConsumer(Schema.INT32)
+ .topic(topic)
+ .subscriptionName("sub1")
+ .readCompacted(true)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe()) {
+ int received = 0;
+ while(true) {
Review Comment:
Updated. It's strange why the style check didn't check it 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312777867
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -443,35 +421,50 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
<T> CompletableFuture<Boolean> addToCompactedLedger(
LedgerHandle lh, Message<T> m, String topic, Semaphore outstanding) {
- CompletableFuture<Boolean> bkf = new CompletableFuture<>();
- if (m == null || batchMessageContainer.add((MessageImpl<?>) m, null)) {
- if (batchMessageContainer.getNumMessagesInBatch() > 0) {
- try {
- ByteBuf serialized = batchMessageContainer.toByteBuf();
- outstanding.acquire();
- mxBean.addCompactionWriteOp(topic, serialized.readableBytes());
- long start = System.nanoTime();
- lh.asyncAddEntry(serialized,
- (rc, ledger, eid, ctx) -> {
- outstanding.release();
- mxBean.addCompactionLatencyOp(topic, System.nanoTime() - start, TimeUnit.NANOSECONDS);
- if (rc != BKException.Code.OK) {
- bkf.completeExceptionally(BKException.create(rc));
- } else {
- bkf.complete(true);
- }
- }, null);
-
- } catch (Throwable t) {
- log.error("Failed to add entry", t);
- batchMessageContainer.discard((Exception) t);
- return FutureUtil.failedFuture(t);
- }
- } else {
- bkf.complete(false);
+ if (m == null) {
+ return flushBatchMessage(lh, topic, outstanding);
+ }
+ if (batchMessageContainer.haveEnoughSpace((MessageImpl<?>) m)) {
+ if (batchMessageContainer.add((MessageImpl<?>) m, null)) {
+ return flushBatchMessage(lh, topic, outstanding);
}
Review Comment:
Should we remove this check? With the current implementation (`maxNumMessagesInBatch` and `maxBytesInBatch` are both `Integer.MAX_VALUE`), `add` never returns true.
--
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] Demogorgon314 commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312776449
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -44,16 +44,17 @@
* [(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]
*/
public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
- MessageCrypto msgCrypto;
- Set<String> encryptionKeys;
- CryptoKeyReader cryptoKeyReader;
+ private MessageCrypto<MessageMetadata, MessageMetadata> msgCrypto;
+ private Set<String> encryptionKeys;
+ private CryptoKeyReader cryptoKeyReader;
+ private MessageIdImpl lastAddedMessageId;
- public RawBatchMessageContainerImpl(int maxNumMessagesInBatch, int maxBytesInBatch) {
+ public RawBatchMessageContainerImpl() {
super();
this.compressionType = CompressionType.NONE;
this.compressor = new CompressionCodecNone();
- this.maxNumMessagesInBatch = maxNumMessagesInBatch;
- this.maxBytesInBatch = maxBytesInBatch;
+ this.maxNumMessagesInBatch = Integer.MAX_VALUE;
+ this.maxBytesInBatch = Integer.MAX_VALUE;
Review Comment:
You can see we called `super.add(msg, callback);`, in this method, it will use the `isBatchFull` method. Override is not working for the parent class.
```
@Override
public boolean add(MessageImpl<?> msg, SendCallback callback) {
this.lastAddedMessageId = (MessageIdAdv) msg.getMessageId();
return super.add(msg, callback);
}
```
--
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] Demogorgon314 commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1311329736
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -418,10 +418,9 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
.whenComplete((res, exception2) -> {
if (exception2 != null) {
promise.completeExceptionally(exception2);
- return;
}
+ phaseTwoLoop(topic, reader, lh, outstanding, promise);
});
- phaseTwoLoop(topic, reader, lh, outstanding, promise);
Review Comment:
The `RawBatchMessageContainerImpl ` is not thread-safe, right? If we add entries parallel, I wonder if this will cause data loss?(add message and clear at same 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] heesung-sn commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1311139619
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -443,32 +442,50 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
<T> CompletableFuture<Boolean> addToCompactedLedger(
LedgerHandle lh, Message<T> m, String topic, Semaphore outstanding) {
+ if (m == null) {
+ return flushBatchMessage(lh, topic, outstanding);
+ }
+ if (batchMessageContainer.haveEnoughSpace((MessageImpl<?>) m)) {
+ if (batchMessageContainer.add((MessageImpl<?>) m, null)) {
+ return flushBatchMessage(lh, topic, outstanding);
+ }
+ return CompletableFuture.completedFuture(false);
+ }
+ return flushBatchMessage(lh, topic, outstanding)
+ .thenCompose(__ -> {
+ if (batchMessageContainer.add((MessageImpl<?>) m, null)) {
+ return flushBatchMessage(lh, topic, outstanding);
+ } else {
+ return CompletableFuture.completedFuture(false);
+ }
+ });
+ }
+
+ private CompletableFuture<Boolean> flushBatchMessage(LedgerHandle lh, String topic,
+ Semaphore outstanding) {
CompletableFuture<Boolean> bkf = new CompletableFuture<>();
- if (m == null || batchMessageContainer.add((MessageImpl<?>) m, null)) {
- if (batchMessageContainer.getNumMessagesInBatch() > 0) {
- try {
- ByteBuf serialized = batchMessageContainer.toByteBuf();
- outstanding.acquire();
- mxBean.addCompactionWriteOp(topic, serialized.readableBytes());
- long start = System.nanoTime();
- lh.asyncAddEntry(serialized,
- (rc, ledger, eid, ctx) -> {
- outstanding.release();
- mxBean.addCompactionLatencyOp(topic, System.nanoTime() - start, TimeUnit.NANOSECONDS);
- if (rc != BKException.Code.OK) {
- bkf.completeExceptionally(BKException.create(rc));
- } else {
- bkf.complete(true);
- }
- }, null);
+ if (batchMessageContainer.getNumMessagesInBatch() > 0) {
+ try {
+ ByteBuf serialized = batchMessageContainer.toByteBuf();
+ outstanding.acquire();
Review Comment:
Nit :Lets move this acquire right before addEntry
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -443,32 +442,50 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
<T> CompletableFuture<Boolean> addToCompactedLedger(
LedgerHandle lh, Message<T> m, String topic, Semaphore outstanding) {
+ if (m == null) {
+ return flushBatchMessage(lh, topic, outstanding);
+ }
+ if (batchMessageContainer.haveEnoughSpace((MessageImpl<?>) m)) {
+ if (batchMessageContainer.add((MessageImpl<?>) m, null)) {
+ return flushBatchMessage(lh, topic, outstanding);
+ }
+ return CompletableFuture.completedFuture(false);
+ }
+ return flushBatchMessage(lh, topic, outstanding)
+ .thenCompose(__ -> {
+ if (batchMessageContainer.add((MessageImpl<?>) m, null)) {
+ return flushBatchMessage(lh, topic, outstanding);
+ } else {
+ return CompletableFuture.completedFuture(false);
+ }
+ });
+ }
+
+ private CompletableFuture<Boolean> flushBatchMessage(LedgerHandle lh, String topic,
+ Semaphore outstanding) {
CompletableFuture<Boolean> bkf = new CompletableFuture<>();
- if (m == null || batchMessageContainer.add((MessageImpl<?>) m, null)) {
- if (batchMessageContainer.getNumMessagesInBatch() > 0) {
- try {
- ByteBuf serialized = batchMessageContainer.toByteBuf();
- outstanding.acquire();
- mxBean.addCompactionWriteOp(topic, serialized.readableBytes());
- long start = System.nanoTime();
- lh.asyncAddEntry(serialized,
- (rc, ledger, eid, ctx) -> {
- outstanding.release();
- mxBean.addCompactionLatencyOp(topic, System.nanoTime() - start, TimeUnit.NANOSECONDS);
- if (rc != BKException.Code.OK) {
- bkf.completeExceptionally(BKException.create(rc));
- } else {
- bkf.complete(true);
- }
- }, null);
+ if (batchMessageContainer.getNumMessagesInBatch() > 0) {
+ try {
+ ByteBuf serialized = batchMessageContainer.toByteBuf();
+ outstanding.acquire();
+ mxBean.addCompactionWriteOp(topic, serialized.readableBytes());
+ long start = System.nanoTime();
+ lh.asyncAddEntry(serialized,
+ (rc, ledger, eid, ctx) -> {
+ outstanding.release();
+ mxBean.addCompactionLatencyOp(topic, System.nanoTime() - start, TimeUnit.NANOSECONDS);
+ if (rc != BKException.Code.OK) {
+ bkf.completeExceptionally(BKException.create(rc));
+ } else {
+ bkf.complete(true);
+ }
+ }, null);
Review Comment:
Nit: lets try to release the semaphore after completing the future.
--
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] codelipenghui merged pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui merged PR #21091:
URL: https://github.com/apache/pulsar/pull/21091
--
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] Demogorgon314 commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312630264
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -90,6 +91,23 @@ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
this.cryptoKeyReader = cryptoKeyReader;
}
+ @Override
+ public boolean add(MessageImpl<?> msg, SendCallback callback) {
+ this.lastAddedMessageId = (MessageIdImpl) msg.getMessageId();
+ return super.add(msg, callback);
+ }
+
+ @Override
+ public boolean haveEnoughSpace(MessageImpl<?> msg) {
+ if (lastAddedMessageId == null) {
+ return true;
+ }
+ // Keep same batch compact to same batch.
+ MessageIdImpl msgId = (MessageIdImpl) msg.getMessageId();
+ return msgId.getLedgerId() == lastAddedMessageId.getLedgerId()
+ && msgId.getEntryId() == lastAddedMessageId.getEntryId();
Review Comment:
I only want to compare the `LedgerId` and `EntryId`. Can the `MessageIdAdvUtils#equals` method do that?
--
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] Demogorgon314 commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1311571071
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -418,10 +418,9 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
.whenComplete((res, exception2) -> {
if (exception2 != null) {
promise.completeExceptionally(exception2);
- return;
}
+ phaseTwoLoop(topic, reader, lh, outstanding, promise);
});
- phaseTwoLoop(topic, reader, lh, outstanding, promise);
Review Comment:
Probably, we should use the raw reader to do this, and we have to compact each batch into different entries.
--
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] heesung-sn commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1311821809
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -418,10 +418,9 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
.whenComplete((res, exception2) -> {
if (exception2 != null) {
promise.completeExceptionally(exception2);
- return;
}
+ phaseTwoLoop(topic, reader, lh, outstanding, promise);
});
- phaseTwoLoop(topic, reader, lh, outstanding, promise);
Review Comment:
> Probably we should use the raw reader to do this, and we have to compact each batch into different entries.
We could do, but we need to get values from the RawReader to apply the compaction strategy. We are using the reader interface to reuse the value-parsing 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Demogorgon314 commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312444892
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -90,6 +91,26 @@ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
this.cryptoKeyReader = cryptoKeyReader;
}
+ @Override
+ public boolean add(MessageImpl<?> msg, SendCallback callback) {
+ this.lastAddedMessageId = (MessageIdImpl) msg.getMessageId();
+ return super.add(msg, callback);
+ }
+
+ @Override
+ public boolean haveEnoughSpace(MessageImpl<?> msg) {
+ if (lastAddedMessageId == null) {
+ return super.haveEnoughSpace(msg);
+ }
+ // Keep same batch compact to same batch.
+ MessageIdImpl msgId = (MessageIdImpl) msg.getMessageId();
+ if (msgId.getLedgerId() != lastAddedMessageId.getLedgerId()
+ || msgId.getEntryId() != lastAddedMessageId.getEntryId()) {
+ return false;
+ }
+ return super.haveEnoughSpace(msg);
Review Comment:
Yes, maybe we can set the value to `Integer.MAX_VALUE`, I will update it.
```
this.maxNumMessagesInBatch = Integer.MAX_VALUE;
this.maxBytesInBatch = Integer.MAX_VALUE;
```
--
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] heesung-sn commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312452272
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java:
##########
@@ -300,6 +301,9 @@ public synchronized void start() throws PulsarServerException {
producer = pulsar.getClient().newProducer(schema)
.enableBatching(true)
+ // Make sure input batch size =< compacted output batch,
+ // Otherwise, the compactor will split the batch to multiple batches.
Review Comment:
nit: Please remove or update this comment, too.
--
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] Demogorgon314 commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312658629
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -377,7 +356,7 @@ private <T> CompletableFuture<Long> runPhaseTwo(
return;
}
});
- })
+ }, scheduler)
Review Comment:
Oh, I changed back. This change is made when I am debugging, this change is no useful.
--
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 a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312797218
##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/StrategicCompactionTest.java:
##########
@@ -148,24 +159,56 @@ public void testNumericOrderCompaction() throws Exception {
Assert.assertEquals(tableView.entrySet(), expectedCopy.entrySet());
}
- @Override
- public void testCompactCompressedBatching() throws Exception {
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 10);
- super.testCompactCompressedBatching();
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 1);
- }
+ @Test(timeOut = 20000)
+ public void testSameBatchCompactToSameBatch() throws Exception {
+ final String topic =
+ "persistent://my-property/use/my-ns/testSameBatchCompactToSameBatch" + UUID.randomUUID();
- @Override
- public void testCompactEncryptedAndCompressedBatching() throws Exception {
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 10);
- super.testCompactEncryptedAndCompressedBatching();
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 1);
- }
+ final int messages = 10;
+
+ // 1.create producer and publish message to the topic.
+ ProducerBuilder<Integer> builder = pulsarClient.newProducer(Schema.INT32)
+ .compressionType(MSG_COMPRESSION_TYPE).topic(topic);
+ builder.batchingMaxMessages(2)
+ .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS);
+
+ Producer<Integer> producer = builder.create();
+
+ List<CompletableFuture<MessageId>> futures = new ArrayList<>(messages);
+ for (int i = 0; i < messages; i++) {
+ futures.add(producer.newMessage().key(String.valueOf(i))
+ .value(i)
+ .sendAsync());
+ }
+
+ FutureUtil.waitForAll(futures).get();
+
+ // 2.compact the topic.
+ StrategicTwoPhaseCompactor compactor
+ = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+ compactor.compact(topic, strategy).get();
+
+ // consumer with readCompacted enabled only get compacted entries
+ try (Consumer<Integer> consumer = pulsarClient
+ .newConsumer(Schema.INT32)
+ .topic(topic)
+ .subscriptionName("sub1")
+ .readCompacted(true)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe()) {
+ int received = 0;
+ while(true) {
+ Message<Integer> m = consumer.receive(2, TimeUnit.SECONDS);
+ if (m == null) {
+ break;
+ }
+ MessageIdImpl messageId = (MessageIdImpl) m.getMessageId();
Review Comment:
Use `MessageIdAdv` 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312771563
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -44,16 +44,17 @@
* [(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]
*/
public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
- MessageCrypto msgCrypto;
- Set<String> encryptionKeys;
- CryptoKeyReader cryptoKeyReader;
+ private MessageCrypto<MessageMetadata, MessageMetadata> msgCrypto;
+ private Set<String> encryptionKeys;
+ private CryptoKeyReader cryptoKeyReader;
+ private MessageIdImpl lastAddedMessageId;
- public RawBatchMessageContainerImpl(int maxNumMessagesInBatch, int maxBytesInBatch) {
+ public RawBatchMessageContainerImpl() {
super();
this.compressionType = CompressionType.NONE;
this.compressor = new CompressionCodecNone();
- this.maxNumMessagesInBatch = maxNumMessagesInBatch;
- this.maxBytesInBatch = maxBytesInBatch;
+ this.maxNumMessagesInBatch = Integer.MAX_VALUE;
+ this.maxBytesInBatch = Integer.MAX_VALUE;
Review Comment:
> if we override it, we need to override the add method as well.
Sorry I didn't get it. I mean, if you made the following changes:
```diff
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java
index a40ab35dab..67df65279b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java
@@ -54,8 +54,6 @@ public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
super();
this.compressionType = CompressionType.NONE;
this.compressor = new CompressionCodecNone();
- this.maxNumMessagesInBatch = Integer.MAX_VALUE;
- this.maxBytesInBatch = Integer.MAX_VALUE;
}
private ByteBuf encrypt(ByteBuf compressedPayload) {
if (msgCrypto == null) {
@@ -98,6 +96,11 @@ public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
return super.add(msg, callback);
}
+ @Override
+ protected boolean isBatchFull() {
+ return false;
+ }
+
@Override
public boolean haveEnoughSpace(MessageImpl<?> msg) {
if (lastAddedMessageId == null) {
```
what else should you make in addition?
--
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 a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312618835
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -44,16 +44,17 @@
* [(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]
*/
public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
- MessageCrypto msgCrypto;
- Set<String> encryptionKeys;
- CryptoKeyReader cryptoKeyReader;
+ private MessageCrypto<MessageMetadata, MessageMetadata> msgCrypto;
+ private Set<String> encryptionKeys;
+ private CryptoKeyReader cryptoKeyReader;
+ private MessageIdImpl lastAddedMessageId;
- public RawBatchMessageContainerImpl(int maxNumMessagesInBatch, int maxBytesInBatch) {
+ public RawBatchMessageContainerImpl() {
super();
this.compressionType = CompressionType.NONE;
this.compressor = new CompressionCodecNone();
- this.maxNumMessagesInBatch = maxNumMessagesInBatch;
- this.maxBytesInBatch = maxBytesInBatch;
+ this.maxNumMessagesInBatch = Integer.MAX_VALUE;
+ this.maxBytesInBatch = Integer.MAX_VALUE;
Review Comment:
It seems that these two fields only affect `haveEnoughSpace` and `isFull`. If your intention is to customize the behavior of these two methods, I think it's better to just override them. Since `hasEnoughSpace` is already overridden, you can override `isFull` with a trivial implementation that returns false.
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -90,6 +91,23 @@ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
this.cryptoKeyReader = cryptoKeyReader;
}
+ @Override
+ public boolean add(MessageImpl<?> msg, SendCallback callback) {
+ this.lastAddedMessageId = (MessageIdImpl) msg.getMessageId();
+ return super.add(msg, callback);
+ }
+
+ @Override
+ public boolean haveEnoughSpace(MessageImpl<?> msg) {
+ if (lastAddedMessageId == null) {
+ return true;
+ }
+ // Keep same batch compact to same batch.
+ MessageIdImpl msgId = (MessageIdImpl) msg.getMessageId();
+ return msgId.getLedgerId() == lastAddedMessageId.getLedgerId()
+ && msgId.getEntryId() == lastAddedMessageId.getEntryId();
Review Comment:
If you just want to compare if `msg.getMessageId()` equals `lastAddedMessageId`, you can use `MessageIdAdvUtils#equals`
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -443,32 +421,48 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
<T> CompletableFuture<Boolean> addToCompactedLedger(
LedgerHandle lh, Message<T> m, String topic, Semaphore outstanding) {
+ if (m == null) {
+ return flushBatchMessage(lh, topic, outstanding);
+ }
+ if (batchMessageContainer.haveEnoughSpace((MessageImpl<?>) m)) {
+ if (batchMessageContainer.add((MessageImpl<?>) m, null)) {
+ return flushBatchMessage(lh, topic, outstanding);
+ }
+ return CompletableFuture.completedFuture(false);
+ }
+ CompletableFuture<Boolean> f = flushBatchMessage(lh, topic, outstanding);
+ if (batchMessageContainer.add((MessageImpl<?>) m, null)) {
+ return flushBatchMessage(lh, topic, outstanding).thenCombine(f, (a, b) -> a && b);
+ } else {
+ return f;
+ }
+ }
+
+ private CompletableFuture<Boolean> flushBatchMessage(LedgerHandle lh, String topic,
+ Semaphore outstanding) {
CompletableFuture<Boolean> bkf = new CompletableFuture<>();
- if (m == null || batchMessageContainer.add((MessageImpl<?>) m, null)) {
- if (batchMessageContainer.getNumMessagesInBatch() > 0) {
- try {
- ByteBuf serialized = batchMessageContainer.toByteBuf();
- outstanding.acquire();
- mxBean.addCompactionWriteOp(topic, serialized.readableBytes());
- long start = System.nanoTime();
- lh.asyncAddEntry(serialized,
- (rc, ledger, eid, ctx) -> {
- outstanding.release();
- mxBean.addCompactionLatencyOp(topic, System.nanoTime() - start, TimeUnit.NANOSECONDS);
- if (rc != BKException.Code.OK) {
- bkf.completeExceptionally(BKException.create(rc));
- } else {
- bkf.complete(true);
- }
- }, null);
+ if (batchMessageContainer.getNumMessagesInBatch() > 0) {
Review Comment:
I see you already made much changes to `addToCompactedLedger`, for the refactored code, you can just use early return.
The previous code:
```java
if (m == null || batchMessageContainer.add((MessageImpl<?>) m, null)) {
if (batchMessageContainer.getNumMessagesInBatch() > 0) {
try {
/* ... */
} catch (Throwable t) {
log.error("Failed to add entry", t);
batchMessageContainer.discard((Exception) t);
return FutureUtil.failedFuture(t);
}
} else {
bkf.complete(false);
}
} else {
bkf.complete(false);
}
```
After applying early return idiom:
```java
if (m != null && !batchMessageContainer.add((MessageImpl<?>) m, null)) {
return CompletableFuture.completedFuture(false);
}
if (batchMessageContainer.getNumMessagesInBatch() <= 0) {
return CompletableFuture.completedFuture(false);
}
// try-catch...
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -90,6 +91,23 @@ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
this.cryptoKeyReader = cryptoKeyReader;
}
+ @Override
+ public boolean add(MessageImpl<?> msg, SendCallback callback) {
+ this.lastAddedMessageId = (MessageIdImpl) msg.getMessageId();
+ return super.add(msg, callback);
+ }
+
+ @Override
+ public boolean haveEnoughSpace(MessageImpl<?> msg) {
+ if (lastAddedMessageId == null) {
Review Comment:
In which case could `lastAddedMessageId` be `null`? What I can think of is when there is no message 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] Demogorgon314 commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312640863
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -44,16 +44,17 @@
* [(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]
*/
public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
- MessageCrypto msgCrypto;
- Set<String> encryptionKeys;
- CryptoKeyReader cryptoKeyReader;
+ private MessageCrypto<MessageMetadata, MessageMetadata> msgCrypto;
+ private Set<String> encryptionKeys;
+ private CryptoKeyReader cryptoKeyReader;
+ private MessageIdImpl lastAddedMessageId;
- public RawBatchMessageContainerImpl(int maxNumMessagesInBatch, int maxBytesInBatch) {
+ public RawBatchMessageContainerImpl() {
super();
this.compressionType = CompressionType.NONE;
this.compressor = new CompressionCodecNone();
- this.maxNumMessagesInBatch = maxNumMessagesInBatch;
- this.maxBytesInBatch = maxBytesInBatch;
+ this.maxNumMessagesInBatch = Integer.MAX_VALUE;
+ this.maxBytesInBatch = Integer.MAX_VALUE;
Review Comment:
The `isBatchFull` method will be used by `org.apache.pulsar.client.impl.BatchMessageContainerImpl#add`, if we override it, we need to override the `add` method 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] heesung-sn commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1311933192
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -90,6 +91,26 @@ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
this.cryptoKeyReader = cryptoKeyReader;
}
+ @Override
+ public boolean add(MessageImpl<?> msg, SendCallback callback) {
+ this.lastAddedMessageId = (MessageIdImpl) msg.getMessageId();
+ return super.add(msg, callback);
+ }
+
+ @Override
+ public boolean haveEnoughSpace(MessageImpl<?> msg) {
+ if (lastAddedMessageId == null) {
+ return super.haveEnoughSpace(msg);
+ }
+ // Keep same batch compact to same batch.
+ MessageIdImpl msgId = (MessageIdImpl) msg.getMessageId();
+ if (msgId.getLedgerId() != lastAddedMessageId.getLedgerId()
+ || msgId.getEntryId() != lastAddedMessageId.getEntryId()) {
+ return false;
+ }
+ return super.haveEnoughSpace(msg);
Review Comment:
Now, I think we should return true here and remove the below member vars. This batching should not be bounded by a specific size now.
```
this.maxNumMessagesInBatch = maxNumMessagesInBatch;
this.maxBytesInBatch = maxBytesInBatch;
```
--
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] Demogorgon314 commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312473319
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -90,6 +91,26 @@ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
this.cryptoKeyReader = cryptoKeyReader;
}
+ @Override
+ public boolean add(MessageImpl<?> msg, SendCallback callback) {
+ this.lastAddedMessageId = (MessageIdImpl) msg.getMessageId();
+ return super.add(msg, callback);
+ }
+
+ @Override
+ public boolean haveEnoughSpace(MessageImpl<?> msg) {
+ if (lastAddedMessageId == null) {
+ return super.haveEnoughSpace(msg);
+ }
+ // Keep same batch compact to same batch.
+ MessageIdImpl msgId = (MessageIdImpl) msg.getMessageId();
+ if (msgId.getLedgerId() != lastAddedMessageId.getLedgerId()
+ || msgId.getEntryId() != lastAddedMessageId.getEntryId()) {
+ return false;
+ }
+ return super.haveEnoughSpace(msg);
Review Comment:
Updated.
--
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 a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312811572
##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/StrategicCompactionTest.java:
##########
@@ -148,24 +160,65 @@ public void testNumericOrderCompaction() throws Exception {
Assert.assertEquals(tableView.entrySet(), expectedCopy.entrySet());
}
- @Override
- public void testCompactCompressedBatching() throws Exception {
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 10);
- super.testCompactCompressedBatching();
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 1);
- }
+ @Test(timeOut = 20000)
+ public void testSameBatchCompactToSameBatch() throws Exception {
+ final String topic =
+ "persistent://my-property/use/my-ns/testSameBatchCompactToSameBatch" + UUID.randomUUID();
- @Override
- public void testCompactEncryptedAndCompressedBatching() throws Exception {
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 10);
- super.testCompactEncryptedAndCompressedBatching();
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 1);
- }
+ final int messages = 10;
+
+ // 1.create producer and publish message to the topic.
+ ProducerBuilder<Integer> builder = pulsarClient.newProducer(Schema.INT32)
+ .compressionType(MSG_COMPRESSION_TYPE).topic(topic);
+ builder.batchingMaxMessages(2)
+ .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS);
+
+ Producer<Integer> producer = builder.create();
+
+ List<CompletableFuture<MessageId>> futures = new ArrayList<>(messages);
+ for (int i = 0; i < messages; i++) {
+ futures.add(producer.newMessage().key(String.valueOf(i))
+ .value(i)
+ .sendAsync());
+ }
+ // Add additional message to make sure the last batch is not full and test flush this batch.
+ futures.add(producer.newMessage().key(String.valueOf(messages))
+ .value(messages)
+ .sendAsync());
+
+ FutureUtil.waitForAll(futures).get();
+
+ // 2.compact the topic.
+ StrategicTwoPhaseCompactor compactor
+ = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+ compactor.compact(topic, strategy).get();
+
+ // consumer with readCompacted enabled only get compacted entries
+ try (Consumer<Integer> consumer = pulsarClient
+ .newConsumer(Schema.INT32)
+ .topic(topic)
+ .subscriptionName("sub1")
+ .readCompacted(true)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe()) {
+ int received = 0;
+ while(true) {
Review Comment:
```suggestion
while (true) {
```
##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/StrategicCompactionTest.java:
##########
@@ -148,24 +160,65 @@ public void testNumericOrderCompaction() throws Exception {
Assert.assertEquals(tableView.entrySet(), expectedCopy.entrySet());
}
- @Override
- public void testCompactCompressedBatching() throws Exception {
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 10);
- super.testCompactCompressedBatching();
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 1);
- }
+ @Test(timeOut = 20000)
+ public void testSameBatchCompactToSameBatch() throws Exception {
+ final String topic =
+ "persistent://my-property/use/my-ns/testSameBatchCompactToSameBatch" + UUID.randomUUID();
- @Override
- public void testCompactEncryptedAndCompressedBatching() throws Exception {
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 10);
- super.testCompactEncryptedAndCompressedBatching();
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 1);
- }
+ final int messages = 10;
+
+ // 1.create producer and publish message to the topic.
+ ProducerBuilder<Integer> builder = pulsarClient.newProducer(Schema.INT32)
+ .compressionType(MSG_COMPRESSION_TYPE).topic(topic);
+ builder.batchingMaxMessages(2)
+ .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS);
+
+ Producer<Integer> producer = builder.create();
+
+ List<CompletableFuture<MessageId>> futures = new ArrayList<>(messages);
+ for (int i = 0; i < messages; i++) {
+ futures.add(producer.newMessage().key(String.valueOf(i))
+ .value(i)
+ .sendAsync());
+ }
+ // Add additional message to make sure the last batch is not full and test flush this batch.
Review Comment:
It seems to be equivalent with
```java
for (int i = 0; i < messages + 1; i++) {
/* ... */
}
```
So why not use use `messages + 1` as the `messages`?
--
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] heesung-sn commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1311933192
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -90,6 +91,26 @@ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
this.cryptoKeyReader = cryptoKeyReader;
}
+ @Override
+ public boolean add(MessageImpl<?> msg, SendCallback callback) {
+ this.lastAddedMessageId = (MessageIdImpl) msg.getMessageId();
+ return super.add(msg, callback);
+ }
+
+ @Override
+ public boolean haveEnoughSpace(MessageImpl<?> msg) {
+ if (lastAddedMessageId == null) {
+ return super.haveEnoughSpace(msg);
+ }
+ // Keep same batch compact to same batch.
+ MessageIdImpl msgId = (MessageIdImpl) msg.getMessageId();
+ if (msgId.getLedgerId() != lastAddedMessageId.getLedgerId()
+ || msgId.getEntryId() != lastAddedMessageId.getEntryId()) {
+ return false;
+ }
+ return super.haveEnoughSpace(msg);
Review Comment:
Now, I think we should return true here and remove the below member vars. This re-batching should not be bounded by a specific size now.
```
this.maxNumMessagesInBatch = maxNumMessagesInBatch;
this.maxBytesInBatch = maxBytesInBatch;
```
--
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] Demogorgon314 commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312799481
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -44,16 +44,17 @@
* [(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]
*/
public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
- MessageCrypto msgCrypto;
- Set<String> encryptionKeys;
- CryptoKeyReader cryptoKeyReader;
+ private MessageCrypto<MessageMetadata, MessageMetadata> msgCrypto;
+ private Set<String> encryptionKeys;
+ private CryptoKeyReader cryptoKeyReader;
+ private MessageIdImpl lastAddedMessageId;
- public RawBatchMessageContainerImpl(int maxNumMessagesInBatch, int maxBytesInBatch) {
+ public RawBatchMessageContainerImpl() {
super();
this.compressionType = CompressionType.NONE;
this.compressor = new CompressionCodecNone();
- this.maxNumMessagesInBatch = maxNumMessagesInBatch;
- this.maxBytesInBatch = maxBytesInBatch;
+ this.maxNumMessagesInBatch = Integer.MAX_VALUE;
+ this.maxBytesInBatch = Integer.MAX_VALUE;
Review Comment:
Oh, yes, you are right.
--
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 a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312824459
##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/StrategicCompactionTest.java:
##########
@@ -148,24 +160,65 @@ public void testNumericOrderCompaction() throws Exception {
Assert.assertEquals(tableView.entrySet(), expectedCopy.entrySet());
}
- @Override
- public void testCompactCompressedBatching() throws Exception {
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 10);
- super.testCompactCompressedBatching();
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 1);
- }
+ @Test(timeOut = 20000)
+ public void testSameBatchCompactToSameBatch() throws Exception {
+ final String topic =
+ "persistent://my-property/use/my-ns/testSameBatchCompactToSameBatch" + UUID.randomUUID();
- @Override
- public void testCompactEncryptedAndCompressedBatching() throws Exception {
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 10);
- super.testCompactEncryptedAndCompressedBatching();
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 1);
- }
+ final int messages = 10;
+
+ // 1.create producer and publish message to the topic.
+ ProducerBuilder<Integer> builder = pulsarClient.newProducer(Schema.INT32)
+ .compressionType(MSG_COMPRESSION_TYPE).topic(topic);
+ builder.batchingMaxMessages(2)
+ .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS);
+
+ Producer<Integer> producer = builder.create();
+
+ List<CompletableFuture<MessageId>> futures = new ArrayList<>(messages);
+ for (int i = 0; i < messages; i++) {
+ futures.add(producer.newMessage().key(String.valueOf(i))
+ .value(i)
+ .sendAsync());
+ }
+ // Add additional message to make sure the last batch is not full and test flush this batch.
Review Comment:
i.e. use an odd number (11) as the `messages`
--
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] heesung-sn commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1311139619
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -443,32 +442,50 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
<T> CompletableFuture<Boolean> addToCompactedLedger(
LedgerHandle lh, Message<T> m, String topic, Semaphore outstanding) {
+ if (m == null) {
+ return flushBatchMessage(lh, topic, outstanding);
+ }
+ if (batchMessageContainer.haveEnoughSpace((MessageImpl<?>) m)) {
+ if (batchMessageContainer.add((MessageImpl<?>) m, null)) {
+ return flushBatchMessage(lh, topic, outstanding);
+ }
+ return CompletableFuture.completedFuture(false);
+ }
+ return flushBatchMessage(lh, topic, outstanding)
+ .thenCompose(__ -> {
+ if (batchMessageContainer.add((MessageImpl<?>) m, null)) {
+ return flushBatchMessage(lh, topic, outstanding);
+ } else {
+ return CompletableFuture.completedFuture(false);
+ }
+ });
+ }
+
+ private CompletableFuture<Boolean> flushBatchMessage(LedgerHandle lh, String topic,
+ Semaphore outstanding) {
CompletableFuture<Boolean> bkf = new CompletableFuture<>();
- if (m == null || batchMessageContainer.add((MessageImpl<?>) m, null)) {
- if (batchMessageContainer.getNumMessagesInBatch() > 0) {
- try {
- ByteBuf serialized = batchMessageContainer.toByteBuf();
- outstanding.acquire();
- mxBean.addCompactionWriteOp(topic, serialized.readableBytes());
- long start = System.nanoTime();
- lh.asyncAddEntry(serialized,
- (rc, ledger, eid, ctx) -> {
- outstanding.release();
- mxBean.addCompactionLatencyOp(topic, System.nanoTime() - start, TimeUnit.NANOSECONDS);
- if (rc != BKException.Code.OK) {
- bkf.completeExceptionally(BKException.create(rc));
- } else {
- bkf.complete(true);
- }
- }, null);
+ if (batchMessageContainer.getNumMessagesInBatch() > 0) {
+ try {
+ ByteBuf serialized = batchMessageContainer.toByteBuf();
+ outstanding.acquire();
Review Comment:
Nit :Lets move this acquire right before addEntry
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -443,32 +442,50 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
<T> CompletableFuture<Boolean> addToCompactedLedger(
LedgerHandle lh, Message<T> m, String topic, Semaphore outstanding) {
+ if (m == null) {
+ return flushBatchMessage(lh, topic, outstanding);
+ }
+ if (batchMessageContainer.haveEnoughSpace((MessageImpl<?>) m)) {
+ if (batchMessageContainer.add((MessageImpl<?>) m, null)) {
+ return flushBatchMessage(lh, topic, outstanding);
+ }
+ return CompletableFuture.completedFuture(false);
+ }
+ return flushBatchMessage(lh, topic, outstanding)
+ .thenCompose(__ -> {
+ if (batchMessageContainer.add((MessageImpl<?>) m, null)) {
+ return flushBatchMessage(lh, topic, outstanding);
+ } else {
+ return CompletableFuture.completedFuture(false);
+ }
+ });
+ }
+
+ private CompletableFuture<Boolean> flushBatchMessage(LedgerHandle lh, String topic,
+ Semaphore outstanding) {
CompletableFuture<Boolean> bkf = new CompletableFuture<>();
- if (m == null || batchMessageContainer.add((MessageImpl<?>) m, null)) {
- if (batchMessageContainer.getNumMessagesInBatch() > 0) {
- try {
- ByteBuf serialized = batchMessageContainer.toByteBuf();
- outstanding.acquire();
- mxBean.addCompactionWriteOp(topic, serialized.readableBytes());
- long start = System.nanoTime();
- lh.asyncAddEntry(serialized,
- (rc, ledger, eid, ctx) -> {
- outstanding.release();
- mxBean.addCompactionLatencyOp(topic, System.nanoTime() - start, TimeUnit.NANOSECONDS);
- if (rc != BKException.Code.OK) {
- bkf.completeExceptionally(BKException.create(rc));
- } else {
- bkf.complete(true);
- }
- }, null);
+ if (batchMessageContainer.getNumMessagesInBatch() > 0) {
+ try {
+ ByteBuf serialized = batchMessageContainer.toByteBuf();
+ outstanding.acquire();
+ mxBean.addCompactionWriteOp(topic, serialized.readableBytes());
+ long start = System.nanoTime();
+ lh.asyncAddEntry(serialized,
+ (rc, ledger, eid, ctx) -> {
+ outstanding.release();
+ mxBean.addCompactionLatencyOp(topic, System.nanoTime() - start, TimeUnit.NANOSECONDS);
+ if (rc != BKException.Code.OK) {
+ bkf.completeExceptionally(BKException.create(rc));
+ } else {
+ bkf.complete(true);
+ }
+ }, null);
Review Comment:
Nit: lets try to release the semaphore after completing the future.
--
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] heesung-sn commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1311129324
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -90,6 +91,26 @@ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
this.cryptoKeyReader = cryptoKeyReader;
}
+ @Override
+ public boolean add(MessageImpl<?> msg, SendCallback callback) {
+ this.lastAddedMessageId = (MessageIdImpl) msg.getMessageId();
Review Comment:
We need to clean this id when we call clear(). I wonder how the added test passed without cleaning this id.
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -443,32 +442,50 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
<T> CompletableFuture<Boolean> addToCompactedLedger(
LedgerHandle lh, Message<T> m, String topic, Semaphore outstanding) {
+ if (m == null) {
+ return flushBatchMessage(lh, topic, outstanding);
+ }
+ if (batchMessageContainer.haveEnoughSpace((MessageImpl<?>) m)) {
+ if (batchMessageContainer.add((MessageImpl<?>) m, null)) {
+ return flushBatchMessage(lh, topic, outstanding);
+ }
+ return CompletableFuture.completedFuture(false);
+ }
+ return flushBatchMessage(lh, topic, outstanding)
+ .thenCompose(__ -> {
+ if (batchMessageContainer.add((MessageImpl<?>) m, null)) {
Review Comment:
Nit: recursively call this func .
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -418,10 +418,9 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
.whenComplete((res, exception2) -> {
if (exception2 != null) {
promise.completeExceptionally(exception2);
- return;
}
+ phaseTwoLoop(topic, reader, lh, outstanding, promise);
});
- phaseTwoLoop(topic, reader, lh, outstanding, promise);
Review Comment:
We want to add entries in parallel . Otherwise , we wont need the outstanding msg semaphore .
--
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 a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312820909
##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/StrategicCompactionTest.java:
##########
@@ -148,24 +160,65 @@ public void testNumericOrderCompaction() throws Exception {
Assert.assertEquals(tableView.entrySet(), expectedCopy.entrySet());
}
- @Override
- public void testCompactCompressedBatching() throws Exception {
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 10);
- super.testCompactCompressedBatching();
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 1);
- }
+ @Test(timeOut = 20000)
+ public void testSameBatchCompactToSameBatch() throws Exception {
+ final String topic =
+ "persistent://my-property/use/my-ns/testSameBatchCompactToSameBatch" + UUID.randomUUID();
- @Override
- public void testCompactEncryptedAndCompressedBatching() throws Exception {
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 10);
- super.testCompactEncryptedAndCompressedBatching();
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 1);
- }
+ final int messages = 10;
+
+ // 1.create producer and publish message to the topic.
+ ProducerBuilder<Integer> builder = pulsarClient.newProducer(Schema.INT32)
+ .compressionType(MSG_COMPRESSION_TYPE).topic(topic);
+ builder.batchingMaxMessages(2)
+ .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS);
+
+ Producer<Integer> producer = builder.create();
+
+ List<CompletableFuture<MessageId>> futures = new ArrayList<>(messages);
+ for (int i = 0; i < messages; i++) {
+ futures.add(producer.newMessage().key(String.valueOf(i))
+ .value(i)
+ .sendAsync());
+ }
+ // Add additional message to make sure the last batch is not full and test flush this batch.
+ futures.add(producer.newMessage().key(String.valueOf(messages))
+ .value(messages)
+ .sendAsync());
+
+ FutureUtil.waitForAll(futures).get();
+
+ // 2.compact the topic.
+ StrategicTwoPhaseCompactor compactor
+ = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+ compactor.compact(topic, strategy).get();
+
+ // consumer with readCompacted enabled only get compacted entries
+ try (Consumer<Integer> consumer = pulsarClient
+ .newConsumer(Schema.INT32)
+ .topic(topic)
+ .subscriptionName("sub1")
+ .readCompacted(true)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe()) {
+ int received = 0;
+ while(true) {
+ Message<Integer> m = consumer.receive(2, TimeUnit.SECONDS);
+ if (m == null) {
+ break;
+ }
+ if (received <= messages - 1) {
+ MessageIdAdv messageId = (MessageIdAdv) m.getMessageId();
+ assertEquals(messageId.getBatchSize(), 2);
+ } else {
+ MessageIdAdv messageId = (MessageIdAdv) m.getMessageId();
+ assertEquals(messageId.getBatchSize(), 0);
+ }
Review Comment:
```suggestion
MessageIdAdv messageId = (MessageIdAdv) m.getMessageId();
if (received <= messages - 1) {
assertEquals(messageId.getBatchSize(), 2);
} else {
assertEquals(messageId.getBatchSize(), 0);
}
```
--
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] heesung-sn commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1311765382
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -418,10 +418,9 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
.whenComplete((res, exception2) -> {
if (exception2 != null) {
promise.completeExceptionally(exception2);
- return;
}
+ phaseTwoLoop(topic, reader, lh, outstanding, promise);
});
- phaseTwoLoop(topic, reader, lh, outstanding, promise);
Review Comment:
We are iterating messages in the single threaded scheduler.
--
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] heesung-sn commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1311765382
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -418,10 +418,9 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
.whenComplete((res, exception2) -> {
if (exception2 != null) {
promise.completeExceptionally(exception2);
- return;
}
+ phaseTwoLoop(topic, reader, lh, outstanding, promise);
});
- phaseTwoLoop(topic, reader, lh, outstanding, promise);
Review Comment:
We are iterating messages in the single thread.
--
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 a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312847496
##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/StrategicCompactionTest.java:
##########
@@ -148,24 +160,65 @@ public void testNumericOrderCompaction() throws Exception {
Assert.assertEquals(tableView.entrySet(), expectedCopy.entrySet());
}
- @Override
- public void testCompactCompressedBatching() throws Exception {
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 10);
- super.testCompactCompressedBatching();
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 1);
- }
+ @Test(timeOut = 20000)
+ public void testSameBatchCompactToSameBatch() throws Exception {
+ final String topic =
+ "persistent://my-property/use/my-ns/testSameBatchCompactToSameBatch" + UUID.randomUUID();
- @Override
- public void testCompactEncryptedAndCompressedBatching() throws Exception {
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 10);
- super.testCompactEncryptedAndCompressedBatching();
- compactor = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler, 1);
- }
+ final int messages = 10;
+
+ // 1.create producer and publish message to the topic.
+ ProducerBuilder<Integer> builder = pulsarClient.newProducer(Schema.INT32)
+ .compressionType(MSG_COMPRESSION_TYPE).topic(topic);
+ builder.batchingMaxMessages(2)
+ .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS);
+
+ Producer<Integer> producer = builder.create();
+
+ List<CompletableFuture<MessageId>> futures = new ArrayList<>(messages);
+ for (int i = 0; i < messages; i++) {
+ futures.add(producer.newMessage().key(String.valueOf(i))
+ .value(i)
+ .sendAsync());
+ }
+ // Add additional message to make sure the last batch is not full and test flush this batch.
+ futures.add(producer.newMessage().key(String.valueOf(messages))
+ .value(messages)
+ .sendAsync());
+
+ FutureUtil.waitForAll(futures).get();
+
+ // 2.compact the topic.
+ StrategicTwoPhaseCompactor compactor
+ = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+ compactor.compact(topic, strategy).get();
+
+ // consumer with readCompacted enabled only get compacted entries
+ try (Consumer<Integer> consumer = pulsarClient
+ .newConsumer(Schema.INT32)
+ .topic(topic)
+ .subscriptionName("sub1")
+ .readCompacted(true)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe()) {
+ int received = 0;
+ while(true) {
Review Comment:
It's because unit tests are never protected by the checkstyle plugin. I just recalled that I found this issue long days ago.
--
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] heesung-sn commented on a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1311129324
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -90,6 +91,26 @@ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
this.cryptoKeyReader = cryptoKeyReader;
}
+ @Override
+ public boolean add(MessageImpl<?> msg, SendCallback callback) {
+ this.lastAddedMessageId = (MessageIdImpl) msg.getMessageId();
Review Comment:
We need to clean this id when we call clear().
--
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 a diff in pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#discussion_r1312643008
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -377,7 +356,7 @@ private <T> CompletableFuture<Long> runPhaseTwo(
return;
}
});
- })
+ }, scheduler)
Review Comment:
Is there any case when `loopPromise` is not completed in the `scheduler`? In `phaseTwoLoop`:
```java
CompletableFuture.runAsync(() -> {
if (reader.hasNext()) {
/* ... */
} else {
promise.complete(null);
return;
}
}, scheduler)
```
`promise.complete(null);` is called in the `scheduler`, I think we can immediately execute the callback of `thenCompose` instead adding the callback again to `scheduler`.
--
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] codecov-commenter commented on pull request #21091: [fix][broker] Avoid splitting one batch message into two entries in StrategicTwoPhaseCompactor
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #21091:
URL: https://github.com/apache/pulsar/pull/21091#issuecomment-1704514850
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/21091?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#21091](https://app.codecov.io/gh/apache/pulsar/pull/21091?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (08571ad) into [master](https://app.codecov.io/gh/apache/pulsar/commit/64d006b3896bcc971a1b0bee6b78a7c33753b92e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (64d006b) will **increase** coverage by `0.15%`.
> Report is 8 commits behind head on master.
> The diff coverage is `76.59%`.
[](https://app.codecov.io/gh/apache/pulsar/pull/21091?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #21091 +/- ##
============================================
+ Coverage 73.06% 73.22% +0.15%
- Complexity 32407 32453 +46
============================================
Files 1887 1887
Lines 139923 139976 +53
Branches 15402 15412 +10
============================================
+ Hits 102239 102500 +261
+ Misses 29608 29368 -240
- Partials 8076 8108 +32
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/21091/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/21091/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `24.19% <10.63%> (?)` | |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/21091/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `25.11% <5.31%> (+0.04%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/21091/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `72.51% <76.59%> (+0.04%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files Changed](https://app.codecov.io/gh/apache/pulsar/pull/21091?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...g/apache/pulsar/compaction/CompactedTopicImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/21091?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbXBhY3Rpb24vQ29tcGFjdGVkVG9waWNJbXBsLmphdmE=) | `60.71% <0.00%> (-3.20%)` | :arrow_down: |
| [.../pulsar/compaction/StrategicTwoPhaseCompactor.java](https://app.codecov.io/gh/apache/pulsar/pull/21091?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbXBhY3Rpb24vU3RyYXRlZ2ljVHdvUGhhc2VDb21wYWN0b3IuamF2YQ==) | `76.69% <74.07%> (+0.19%)` | :arrow_up: |
| [.../apache/pulsar/compaction/CompactedTopicUtils.java](https://app.codecov.io/gh/apache/pulsar/pull/21091?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbXBhY3Rpb24vQ29tcGFjdGVkVG9waWNVdGlscy5qYXZh) | `74.46% <80.00%> (+4.46%)` | :arrow_up: |
| [...che/bookkeeper/mledger/impl/ManagedLedgerImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/21091?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9pbXBsL01hbmFnZWRMZWRnZXJJbXBsLmphdmE=) | `81.35% <83.33%> (+0.01%)` | :arrow_up: |
| [...che/bookkeeper/mledger/impl/ManagedCursorImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/21091?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9pbXBsL01hbmFnZWRDdXJzb3JJbXBsLmphdmE=) | `79.41% <88.46%> (+0.37%)` | :arrow_up: |
| [...lsar/client/impl/RawBatchMessageContainerImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/21091?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1Jhd0JhdGNoTWVzc2FnZUNvbnRhaW5lckltcGwuamF2YQ==) | `93.42% <100.00%> (+0.88%)` | :arrow_up: |
... and [81 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/21091/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
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