You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/07/04 10:12:26 UTC
[pulsar] branch branch-2.7 updated (701fcfa -> dd6e247)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a change to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from 701fcfa Fix failing test. (#11186)
new 498939f fix compaction entry read exception (#11175)
new f19f41f Make the compaction phase one loop timeout configurable (#11206)
new dd6e247 Remove unused listener to reduce creating executor pool. (#11215)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
conf/broker.conf | 4 ++
.../apache/pulsar/broker/ServiceConfiguration.java | 7 +++
.../pulsar/compaction/CompactedTopicImpl.java | 1 +
.../pulsar/compaction/TwoPhaseCompactor.java | 9 +++-
.../pulsar/compaction/CompactedTopicTest.java | 55 ++++++++++++++++++++++
.../apache/pulsar/compaction/CompactorTest.java | 13 +++++
.../impl/FileSystemManagedLedgerOffloader.java | 3 +-
7 files changed, 89 insertions(+), 3 deletions(-)
[pulsar] 01/03: fix compaction entry read exception (#11175)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 498939f404d5727239fae96255117367759078ac
Author: hangc0276 <ch...@apache.org>
AuthorDate: Fri Jul 2 21:34:10 2021 +0800
fix compaction entry read exception (#11175)
(cherry picked from commit c716495a4282093bd81f9cd43f909811213bfb32)
---
.../pulsar/compaction/CompactedTopicImpl.java | 1 +
.../pulsar/compaction/CompactedTopicTest.java | 55 ++++++++++++++++++++++
2 files changed, 56 insertions(+)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
index d30098a..aa7274b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
@@ -108,6 +108,7 @@ public class CompactedTopicImpl implements CompactedTopic {
if (startPoint == NEWER_THAN_COMPACTED) {
cursor.seek(compactionHorizon.getNext());
callback.readEntriesComplete(Collections.emptyList(), ctx);
+ return CompletableFuture.completedFuture(null);
}
return readEntries(context.ledger, startPoint, endPoint)
.thenAccept((entries) -> {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
index e804d11..58be6d8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
@@ -46,10 +46,13 @@ import org.apache.commons.lang3.tuple.Triple;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.client.admin.LongRunningProcessStatus;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.RawMessage;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.RawMessageImpl;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData;
import org.apache.pulsar.common.policies.data.ClusterData;
@@ -306,4 +309,56 @@ public class CompactedTopicTest extends MockedPulsarServiceBaseTest {
producer.close();
}
+
+ @Test(timeOut = 30000)
+ public void testReadMessageFromCompactedLedger() throws Exception {
+ final String key = "1";
+ String msg = "test compaction msg";
+ final String topic = "persistent://my-property/use/my-ns/testCompactWithEmptyMessage-" + UUID.randomUUID();
+ admin.topics().createPartitionedTopic(topic, 1);
+ final int numMessages = 10;
+
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).enableBatching(false).create();
+ for (int i = 0; i < numMessages; ++i) {
+ producer.newMessage().key(key).value(msg).send();
+ }
+
+ admin.topics().triggerCompaction(topic);
+ boolean succeed = retryStrategically((test) -> {
+ try {
+ return LongRunningProcessStatus.Status.SUCCESS.equals(admin.topics().compactionStatus(topic).status);
+ } catch (PulsarAdminException e) {
+ return false;
+ }
+ }, 10, 200);
+
+ Assert.assertTrue(succeed);
+
+
+ final String newKey = "2";
+ String newMsg = "test compaction msg v2";
+ for (int i = 0; i < numMessages; ++i) {
+ producer.newMessage().key(newKey).value(newMsg).send();
+ }
+
+ Reader<String> reader = pulsarClient.newReader(Schema.STRING)
+ .topic(topic)
+ .readCompacted(true)
+ .startMessageId(MessageId.earliest)
+ .create();
+
+ int compactedMsgCount = 0;
+ int nonCompactedMsgCount = 0;
+ while (reader.hasMessageAvailable()) {
+ Message<String> message = reader.readNext();
+ if (key.equals(message.getKey()) && msg.equals(message.getValue())) {
+ compactedMsgCount++;
+ } else if (newKey.equals(message.getKey()) && newMsg.equals(message.getValue())) {
+ nonCompactedMsgCount++;
+ }
+ }
+
+ Assert.assertEquals(compactedMsgCount, 1);
+ Assert.assertEquals(nonCompactedMsgCount, numMessages);
+ }
}
[pulsar] 03/03: Remove unused listener to reduce creating executor
pool. (#11215)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit dd6e247772d5bc0e73a71cc9a40b0ac90e747fc8
Author: Tboy <te...@apache.org>
AuthorDate: Sun Jul 4 16:22:50 2021 +0800
Remove unused listener to reduce creating executor pool. (#11215)
(cherry picked from commit c1c0e9085a4c4e57d83e9ab67d53163afe48f3d4)
---
.../offload/filesystem/impl/FileSystemManagedLedgerOffloader.java | 3 +--
1 file changed, 1 insertion(+), 2 deletions(-)
diff --git a/tiered-storage/file-system/src/main/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemManagedLedgerOffloader.java b/tiered-storage/file-system/src/main/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemManagedLedgerOffloader.java
index 5438459..bb4d833 100644
--- a/tiered-storage/file-system/src/main/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemManagedLedgerOffloader.java
+++ b/tiered-storage/file-system/src/main/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemManagedLedgerOffloader.java
@@ -203,8 +203,7 @@ public class FileSystemManagedLedgerOffloader implements LedgerOffloader {
semaphore.acquire();
countDownLatch = new CountDownLatch(1);
assignmentScheduler.chooseThread(ledgerId).submit(FileSystemWriter.create(ledgerEntriesOnce, dataWriter, semaphore,
- countDownLatch, haveOffloadEntryNumber, this)).addListener(() -> {
- }, Executors.newSingleThreadExecutor());
+ countDownLatch, haveOffloadEntryNumber, this));
needToOffloadFirstEntryNumber = end + 1;
} while (needToOffloadFirstEntryNumber - 1 != readHandle.getLastAddConfirmed() && fileSystemWriteException == null);
countDownLatch.await();
[pulsar] 02/03: Make the compaction phase one loop timeout
configurable (#11206)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f19f41ffcf25ac6e78e329e84d086208ce1f0e8b
Author: lipenghui <pe...@apache.org>
AuthorDate: Sun Jul 4 08:48:43 2021 +0800
Make the compaction phase one loop timeout configurable (#11206)
(cherry picked from commit 96e4fd69e5376fa2924517b045d02f8229679b92)
---
conf/broker.conf | 4 ++++
.../java/org/apache/pulsar/broker/ServiceConfiguration.java | 7 +++++++
.../org/apache/pulsar/compaction/TwoPhaseCompactor.java | 9 ++++++++-
.../java/org/apache/pulsar/compaction/CompactorTest.java | 13 +++++++++++++
4 files changed, 32 insertions(+), 1 deletion(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index f286961..0cd4e1e 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -397,6 +397,10 @@ maxMessageSize=5242880
# Interval between checks to see if topics with compaction policies need to be compacted
brokerServiceCompactionMonitorIntervalInSeconds=60
+# Timeout for the compaction phase one loop.
+# If the execution time of the compaction phase one loop exceeds this time, the compaction will not proceed.
+brokerServiceCompactionPhaseOneLoopTimeInSeconds=30
+
# Whether to enable the delayed delivery for messages.
# If disabled, messages will be immediately delivered and there will
# be no tracking overhead.
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index efd794b..6566ff0 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -1735,6 +1735,13 @@ public class ServiceConfiguration implements PulsarConfiguration {
private int brokerServiceCompactionMonitorIntervalInSeconds = 60;
@FieldContext(
+ category = CATEGORY_SERVER,
+ doc = "Timeout for the compaction phase one loop, If the execution time of the compaction " +
+ "phase one loop exceeds this time, the compaction will not proceed."
+ )
+ private long brokerServiceCompactionPhaseOneLoopTimeInSeconds = 30;
+
+ @FieldContext(
category = CATEGORY_SCHEMA,
doc = "Enforce schema validation on following cases:\n\n"
+ " - if a producer without a schema attempts to produce to a topic with schema, the producer will be\n"
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
index cd1aaee..7d49926 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.compaction;
import com.google.common.collect.ImmutableMap;
import io.netty.buffer.ByteBuf;
import java.io.IOException;
+import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
@@ -62,12 +63,14 @@ public class TwoPhaseCompactor extends Compactor {
private static final Logger log = LoggerFactory.getLogger(TwoPhaseCompactor.class);
private static final int MAX_OUTSTANDING = 500;
private static final String COMPACTED_TOPIC_LEDGER_PROPERTY = "CompactedTopicLedger";
+ private final Duration phaseOneLoopReadTimeout;
public TwoPhaseCompactor(ServiceConfiguration conf,
PulsarClient pulsar,
BookKeeper bk,
ScheduledExecutorService scheduler) {
super(conf, pulsar, bk, scheduler);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
}
@Override
@@ -175,7 +178,7 @@ public class TwoPhaseCompactor extends Compactor {
private void scheduleTimeout(CompletableFuture<RawMessage> future) {
Future<?> timeout = scheduler.schedule(() -> {
future.completeExceptionally(new TimeoutException("Timeout"));
- }, 10, TimeUnit.SECONDS);
+ }, phaseOneLoopReadTimeout.getSeconds(), TimeUnit.SECONDS);
future.whenComplete((res, exception) -> {
timeout.cancel(true);
});
@@ -410,4 +413,8 @@ public class TwoPhaseCompactor extends Compactor {
this.latestForKey = latestForKey;
}
}
+
+ public long getPhaseOneLoopReadTimeoutInSeconds() {
+ return phaseOneLoopReadTimeout.getSeconds();
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java
index 5c5edd4..b5caecc 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java
@@ -39,14 +39,17 @@ import java.util.concurrent.ScheduledExecutorService;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.LedgerEntry;
import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.RawMessage;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.RawMessageImpl;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -219,6 +222,16 @@ public class CompactorTest extends MockedPulsarServiceBaseTest {
compactor.compact(topic).get();
}
+ @Test
+ public void testPhaseOneLoopTimeConfiguration() {
+ ServiceConfiguration configuration = new ServiceConfiguration();
+ configuration.setBrokerServiceCompactionPhaseOneLoopTimeInSeconds(60);
+ TwoPhaseCompactor compactor = new TwoPhaseCompactor(configuration, Mockito.mock(PulsarClientImpl.class),
+ Mockito.mock(BookKeeper.class), compactionScheduler);
+ Assert.assertEquals(compactor.getPhaseOneLoopReadTimeoutInSeconds(), 60);
+
+ }
+
public ByteBuf extractPayload(RawMessage m) throws Exception {
ByteBuf payloadAndMetadata = m.getHeadersAndPayload();
Commands.skipChecksumIfPresent(payloadAndMetadata);