You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2018/01/29 20:01:21 UTC
[incubator-pulsar] branch master updated: Compact algo (#1044)
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new a499850 Compact algo (#1044)
a499850 is described below
commit a499850b3415add489d702bd61f92288b4f41f60
Author: Ivan Kelly <iv...@apache.org>
AuthorDate: Mon Jan 29 21:01:19 2018 +0100
Compact algo (#1044)
* Compaction algorithm for Pulsar
Compaction will go through the topic in two passes. The first pass
selects latest offset for each key in the topic. Then the second pass
writes these values to a ledger.
The two passes are required to avoid holding the payloads of each of
the latest values in memory, as the payload can be many orders of
magnitude larger than a message id.
* fix nits
* fixup for compactor, empty test
* Shade plugin doesn't like CompletableFuture<?>, which is weird
* Address @rdhabalia comments
* Fix stupid compile error
---
.../org/apache/pulsar/client/api/RawMessage.java | 6 +-
.../apache/pulsar/client/impl/RawMessageImpl.java | 16 +-
.../org/apache/pulsar/compaction/Compactor.java | 86 +++++++
.../pulsar/compaction/TwoPhaseCompactor.java | 285 +++++++++++++++++++++
.../pulsar/client/impl/RawMessageSerDeserTest.java | 2 +-
.../apache/pulsar/client/impl/RawReaderTest.java | 2 +-
.../apache/pulsar/compaction/CompactorTest.java | 223 ++++++++++++++++
7 files changed, 609 insertions(+), 11 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawMessage.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawMessage.java
index cc983e8..e6d3081 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawMessage.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawMessage.java
@@ -46,10 +46,10 @@ public interface RawMessage extends AutoCloseable {
ByteBuf getHeadersAndPayload();
/**
- * Serialize a raw message to a ByteBufPair. The caller is responsible for releasing
- * the returned ByteBufPair.
+ * Serialize a raw message to a ByteBuf. The caller is responsible for releasing
+ * the returned ByteBuf.
*/
- ByteBufPair serialize();
+ ByteBuf serialize();
@Override
void close();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawMessageImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawMessageImpl.java
index 7160cc3..d1b6e14 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawMessageImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawMessageImpl.java
@@ -65,15 +65,18 @@ public class RawMessageImpl implements RawMessage {
}
@Override
- public ByteBufPair serialize() {
+ public ByteBuf serialize() {
+ ByteBuf headersAndPayload = this.headersAndPayload.slice();
+
// Format: [IdSize][Id][PayloadAndMetadataSize][PayloadAndMetadata]
int idSize = id.getSerializedSize();
int headerSize = 4 /* IdSize */ + idSize + 4 /* PayloadAndMetadataSize */;
+ int totalSize = headerSize + headersAndPayload.readableBytes();
- ByteBuf headers = PooledByteBufAllocator.DEFAULT.buffer(headerSize);
- headers.writeInt(idSize);
+ ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(totalSize);
+ buf.writeInt(idSize);
try {
- ByteBufCodedOutputStream outStream = ByteBufCodedOutputStream.get(headers);
+ ByteBufCodedOutputStream outStream = ByteBufCodedOutputStream.get(buf);
id.writeTo(outStream);
outStream.recycle();
} catch (IOException e) {
@@ -81,9 +84,10 @@ public class RawMessageImpl implements RawMessage {
log.error("IO exception serializing to ByteBuf (this shouldn't happen as operation is in-memory)", e);
throw new RuntimeException(e);
}
- headers.writeInt(headersAndPayload.readableBytes());
+ buf.writeInt(headersAndPayload.readableBytes());
+ buf.writeBytes(headersAndPayload);
- return ByteBufPair.get(headers, headersAndPayload);
+ return buf;
}
static public RawMessage deserializeFrom(ByteBuf buffer) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/Compactor.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/Compactor.java
new file mode 100644
index 0000000..62e5fcd
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/Compactor.java
@@ -0,0 +1,86 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.compaction;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+
+import org.apache.bookkeeper.client.BookKeeper;
+
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.RawReader;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compactor for Pulsar topics
+*/
+public abstract class Compactor {
+ private static final Logger log = LoggerFactory.getLogger(Compactor.class);
+ private static final String COMPACTION_SUBSCRIPTION = "__compaction";
+ private static final String COMPACTED_TOPIC_LEDGER_PROPERTY = "CompactedTopicLedger";
+ static BookKeeper.DigestType COMPACTED_TOPIC_LEDGER_DIGEST_TYPE = BookKeeper.DigestType.CRC32;
+ static byte[] COMPACTED_TOPIC_LEDGER_PASSWORD = "".getBytes(UTF_8);
+
+ protected final ServiceConfiguration conf;
+ protected final ScheduledExecutorService scheduler;
+ private final PulsarClient pulsar;
+ private final BookKeeper bk;
+
+ public Compactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this.conf = conf;
+ this.scheduler = scheduler;
+ this.pulsar = pulsar;
+ this.bk = bk;
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ return RawReader.create(pulsar, topic, COMPACTION_SUBSCRIPTION).thenComposeAsync(
+ this::compactAndCloseReader, scheduler);
+ }
+
+ private CompletableFuture<Long> compactAndCloseReader(RawReader reader) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ doCompaction(reader, bk).whenComplete(
+ (ledgerId, exception) -> {
+ reader.closeAsync().whenComplete((v, exception2) -> {
+ if (exception2 != null) {
+ log.warn("Error closing reader handle {}, ignoring", reader, exception2);
+ }
+ if (exception != null) {
+ // complete with original exception
+ promise.completeExceptionally(exception);
+ } else {
+ promise.complete(ledgerId);
+ }
+ });
+ });
+ return promise;
+ }
+
+ protected abstract CompletableFuture<Long> doCompaction(RawReader reader, BookKeeper bk);
+}
+
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
new file mode 100644
index 0000000..957e197
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
@@ -0,0 +1,285 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.compaction;
+
+import com.google.common.collect.ImmutableMap;
+import io.netty.buffer.ByteBuf;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.api.Commands;
+import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.RawReader;
+import org.apache.pulsar.client.api.RawMessage;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects latest offset for each key in the topic. Then the second pass
+ * writes these values to a ledger.
+ *
+ * <p>The two passes are required to avoid holding the payloads of each of
+ * the latest values in memory, as the payload can be many orders of
+ * magnitude larger than a message id.
+*/
+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";
+
+ public TwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ super(conf, pulsar, bk, scheduler);
+ }
+
+ @Override
+ protected CompletableFuture<Long> doCompaction(RawReader reader, BookKeeper bk) {
+ return phaseOne(reader).thenCompose(
+ (r) -> phaseTwo(reader, r.from, r.to, r.latestForKey, bk));
+ }
+
+ private CompletableFuture<PhaseOneResult> phaseOne(RawReader reader) {
+ Map<String,MessageId> latestForKey = new HashMap<>();
+
+ CompletableFuture<PhaseOneResult> loopPromise = new CompletableFuture<>();
+ phaseOneLoop(reader, Optional.empty(), Optional.empty(), latestForKey, loopPromise);
+ return loopPromise;
+ }
+
+ private void phaseOneLoop(RawReader reader,
+ Optional<MessageId> firstMessageId,
+ Optional<MessageId> lastMessageId,
+ Map<String,MessageId> latestForKey,
+ CompletableFuture<PhaseOneResult> loopPromise) {
+ if (loopPromise.isDone()) {
+ return;
+ }
+ CompletableFuture<RawMessage> future = reader.readNextAsync();
+ scheduleTimeout(future);
+ future.whenComplete(
+ (m, exception) -> {
+ try {
+ if (exception != null) {
+ if (exception instanceof TimeoutException
+ && firstMessageId.isPresent()) {
+ loopPromise.complete(new PhaseOneResult(firstMessageId.get(),
+ lastMessageId.get(),
+ latestForKey));
+ } else {
+ loopPromise.completeExceptionally(exception);
+ }
+ return;
+ }
+
+ MessageId id = m.getMessageId();
+ String key = extractKey(m);
+ latestForKey.put(key, id);
+
+ phaseOneLoop(reader,
+ Optional.of(firstMessageId.orElse(id)),
+ Optional.of(id),
+ latestForKey, loopPromise);
+ } finally {
+ m.close();
+ }
+ });
+
+ }
+
+ private void scheduleTimeout(CompletableFuture<RawMessage> future) {
+ Future<?> timeout = scheduler.schedule(() -> {
+ future.completeExceptionally(new TimeoutException("Timeout"));
+ }, 10, TimeUnit.SECONDS);
+ future.whenComplete((res, exception) -> {
+ timeout.cancel(true);
+ });
+ }
+
+ private CompletableFuture<Long> phaseTwo(RawReader reader, MessageId from, MessageId to,
+ Map<String,MessageId> latestForKey, BookKeeper bk) {
+ return createLedger(bk).thenCompose(
+ (ledger) -> phaseTwoSeekThenLoop(reader, from, to, latestForKey, bk, ledger));
+ }
+
+ private CompletableFuture<Long> phaseTwoSeekThenLoop(RawReader reader, MessageId from, MessageId to,
+ Map<String, MessageId> latestForKey,
+ BookKeeper bk, LedgerHandle ledger) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+
+ reader.seekAsync(from).thenCompose((v) -> {
+ Semaphore outstanding = new Semaphore(MAX_OUTSTANDING);
+ CompletableFuture<Void> loopPromise = new CompletableFuture<Void>();
+ phaseTwoLoop(reader, to, latestForKey, ledger, outstanding, loopPromise);
+ return loopPromise;
+ }).thenCompose((v) -> closeLedger(ledger))
+ .thenCompose((v) -> reader.acknowledgeCumulativeAsync(
+ to, ImmutableMap.of(COMPACTED_TOPIC_LEDGER_PROPERTY, ledger.getId())))
+ .whenComplete((res, exception) -> {
+ if (exception != null) {
+ deleteLedger(bk, ledger)
+ .whenComplete((res2, exception2) -> {
+ if (exception2 != null) {
+ log.warn("Cleanup of ledger {} for failed", ledger, exception2);
+ }
+ // complete with original exception
+ promise.completeExceptionally(exception);
+ });
+ } else {
+ promise.complete(ledger.getId());
+ }
+ });
+ return promise;
+ }
+
+ private void phaseTwoLoop(RawReader reader, MessageId to, Map<String, MessageId> latestForKey,
+ LedgerHandle lh, Semaphore outstanding, CompletableFuture<Void> promise) {
+ reader.readNextAsync().whenComplete(
+ (m, exception) -> {
+ try {
+ if (exception != null) {
+ promise.completeExceptionally(exception);
+ return;
+ } else if (promise.isDone()) {
+ return;
+ }
+ MessageId id = m.getMessageId();
+ String key = extractKey(m);
+
+ if (latestForKey.get(key).equals(id)) {
+
+ outstanding.acquire();
+ CompletableFuture<Void> addFuture = addToCompactedLedger(lh, m)
+ .whenComplete((res, exception2) -> {
+ outstanding.release();
+ if (exception2 != null) {
+ promise.completeExceptionally(exception2);
+ }
+ });
+ if (to.equals(id)) {
+ addFuture.whenComplete((res, exception2) -> {
+ if (exception2 == null) {
+ promise.complete(null);
+ }
+ });
+ }
+ }
+ phaseTwoLoop(reader, to, latestForKey, lh, outstanding, promise);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ promise.completeExceptionally(ie);
+ } finally {
+ m.close();
+ }
+ });
+ }
+
+ private CompletableFuture<LedgerHandle> createLedger(BookKeeper bk) {
+ CompletableFuture<LedgerHandle> bkf = new CompletableFuture<>();
+ bk.asyncCreateLedger(conf.getManagedLedgerDefaultEnsembleSize(),
+ conf.getManagedLedgerDefaultWriteQuorum(),
+ conf.getManagedLedgerDefaultAckQuorum(),
+ Compactor.COMPACTED_TOPIC_LEDGER_DIGEST_TYPE,
+ Compactor.COMPACTED_TOPIC_LEDGER_PASSWORD,
+ (rc, ledger, ctx) -> {
+ if (rc != BKException.Code.OK) {
+ bkf.completeExceptionally(BKException.create(rc));
+ } else {
+ bkf.complete(ledger);
+ }
+ }, null);
+ return bkf;
+ }
+
+ private CompletableFuture<Void> deleteLedger(BookKeeper bk, LedgerHandle lh) {
+ CompletableFuture<Void> bkf = new CompletableFuture<>();
+ bk.asyncDeleteLedger(lh.getId(),
+ (rc, ctx) -> {
+ if (rc != BKException.Code.OK) {
+ bkf.completeExceptionally(BKException.create(rc));
+ } else {
+ bkf.complete(null);
+ }
+ }, null);
+ return bkf;
+ }
+
+ private CompletableFuture<Void> closeLedger(LedgerHandle lh) {
+ CompletableFuture<Void> bkf = new CompletableFuture<>();
+ lh.asyncClose((rc, ledger, ctx) -> {
+ if (rc != BKException.Code.OK) {
+ bkf.completeExceptionally(BKException.create(rc));
+ } else {
+ bkf.complete(null);
+ }
+ }, null);
+ return bkf;
+ }
+
+ private CompletableFuture<Void> addToCompactedLedger(LedgerHandle lh, RawMessage m) {
+ CompletableFuture<Void> bkf = new CompletableFuture<>();
+ ByteBuf serialized = m.serialize();
+ lh.asyncAddEntry(serialized,
+ (rc, ledger, eid, ctx) -> {
+ if (rc != BKException.Code.OK) {
+ bkf.completeExceptionally(BKException.create(rc));
+ } else {
+ bkf.complete(null);
+ }
+ }, null);
+ serialized.release();
+ return bkf;
+ }
+
+ private static String extractKey(RawMessage m) {
+ ByteBuf headersAndPayload = m.getHeadersAndPayload();
+ MessageMetadata msgMetadata = Commands.parseMessageMetadata(headersAndPayload);
+ return msgMetadata.getPartitionKey();
+ }
+
+ private static class PhaseOneResult {
+ final MessageId from;
+ final MessageId to;
+ final Map<String,MessageId> latestForKey;
+
+ PhaseOneResult(MessageId from, MessageId to, Map<String,MessageId> latestForKey) {
+ this.from = from;
+ this.to = to;
+ this.latestForKey = latestForKey;
+ }
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawMessageSerDeserTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawMessageSerDeserTest.java
index 035bbf9..79ed794 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawMessageSerDeserTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawMessageSerDeserTest.java
@@ -44,7 +44,7 @@ public class RawMessageSerDeserTest {
.setPartition(10).setBatchIndex(20).build();
RawMessage m = new RawMessageImpl(id, headersAndPayload);
- ByteBuf serialized = ByteBufPair.coalesce(m.serialize());
+ ByteBuf serialized = m.serialize();
byte[] bytes = new byte[serialized.readableBytes()];
serialized.readBytes(bytes);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java
index cef3e42..19cdc31 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java
@@ -116,7 +116,7 @@ public class RawReaderTest extends MockedPulsarServiceBaseTest {
return publishMessagesBase(topic, count, true);
}
- private static String extractKey(RawMessage m) throws Exception {
+ public static String extractKey(RawMessage m) throws Exception {
ByteBuf headersAndPayload = m.getHeadersAndPayload();
MessageMetadata msgMetadata = Commands.parseMessageMetadata(headersAndPayload);
return msgMetadata.getPartitionKey();
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
new file mode 100644
index 0000000..b517a90
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java
@@ -0,0 +1,223 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.compaction;
+
+import static org.apache.pulsar.client.impl.RawReaderTest.extractKey;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.netty.buffer.ByteBuf;
+
+import java.util.ArrayList;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ExecutionException;
+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.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.common.api.Commands;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.PropertyAdmin;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConfiguration;
+import org.apache.pulsar.client.api.MessageBuilder;
+import org.apache.pulsar.client.api.RawMessage;
+import org.apache.pulsar.client.impl.RawMessageImpl;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class CompactorTest extends MockedPulsarServiceBaseTest {
+ private static final Logger log = LoggerFactory.getLogger(CompactorTest.class);
+
+ private ScheduledExecutorService compactionScheduler;
+
+ @BeforeMethod
+ @Override
+ public void setup() throws Exception {
+ super.internalSetup();
+
+ admin.clusters().createCluster("use",
+ new ClusterData("http://127.0.0.1:" + BROKER_WEBSERVICE_PORT));
+ admin.properties().createProperty("my-property",
+ new PropertyAdmin(Lists.newArrayList("appid1", "appid2"), Sets.newHashSet("use")));
+ admin.namespaces().createNamespace("my-property/use/my-ns");
+
+ compactionScheduler = Executors.newSingleThreadScheduledExecutor(
+ new ThreadFactoryBuilder().setNameFormat("compactor").setDaemon(true).build());
+ }
+
+ @AfterMethod
+ @Override
+ public void cleanup() throws Exception {
+ super.internalCleanup();
+
+ compactionScheduler.shutdownNow();
+ }
+
+ private List<String> compactAndVerify(String topic, Map<String, byte[]> expected) throws Exception {
+ BookKeeper bk = pulsar.getBookKeeperClientFactory().create(
+ this.conf, null);
+ Compactor compactor = new TwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+ long compactedLedgerId = compactor.compact(topic).get();
+
+ LedgerHandle ledger = bk.openLedger(compactedLedgerId,
+ Compactor.COMPACTED_TOPIC_LEDGER_DIGEST_TYPE,
+ Compactor.COMPACTED_TOPIC_LEDGER_PASSWORD);
+ Assert.assertEquals(ledger.getLastAddConfirmed() + 1, // 0..lac
+ expected.size(),
+ "Should have as many entries as there is keys");
+
+ List<String> keys = new ArrayList<>();
+ Enumeration<LedgerEntry> entries = ledger.readEntries(0, ledger.getLastAddConfirmed());
+ while (entries.hasMoreElements()) {
+ ByteBuf buf = entries.nextElement().getEntryBuffer();
+ RawMessage m = RawMessageImpl.deserializeFrom(buf);
+ String key = extractKey(m);
+ keys.add(key);
+
+ ByteBuf payload = extractPayload(m);
+ byte[] bytes = new byte[payload.readableBytes()];
+ payload.readBytes(bytes);
+ Assert.assertEquals(bytes, expected.remove(key),
+ "Compacted version should match expected version");
+ m.close();
+ }
+ Assert.assertTrue(expected.isEmpty(), "All expected keys should have been found");
+ return keys;
+ }
+
+ @Test
+ public void testCompaction() throws Exception {
+ String topic = "persistent://my-property/use/my-ns/my-topic1";
+ final int numMessages = 1000;
+ final int maxKeys = 10;
+
+ ProducerConfiguration producerConf = new ProducerConfiguration();
+ Producer producer = pulsarClient.createProducer(topic, producerConf);
+
+ Map<String, byte[]> expected = new HashMap<>();
+ Random r = new Random(0);
+
+ for (int j = 0; j < numMessages; j++) {
+ int keyIndex = r.nextInt(maxKeys);
+ String key = "key"+keyIndex;
+ byte[] data = ("my-message-" + key + "-" + j).getBytes();
+ producer.send(MessageBuilder.create()
+ .setKey(key)
+ .setContent(data).build());
+ expected.put(key, data);
+ }
+ compactAndVerify(topic, expected);
+ }
+
+ @Test
+ public void testCompactAddCompact() throws Exception {
+ String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+ ProducerConfiguration producerConf = new ProducerConfiguration();
+ Producer producer = pulsarClient.createProducer(topic, producerConf);
+
+ Map<String, byte[]> expected = new HashMap<>();
+
+ producer.send(MessageBuilder.create()
+ .setKey("a")
+ .setContent("A_1".getBytes()).build());
+ producer.send(MessageBuilder.create()
+ .setKey("b")
+ .setContent("B_1".getBytes()).build());
+ producer.send(MessageBuilder.create()
+ .setKey("a")
+ .setContent("A_2".getBytes()).build());
+ expected.put("a", "A_2".getBytes());
+ expected.put("b", "B_1".getBytes());
+
+ compactAndVerify(topic, new HashMap<>(expected));
+
+ producer.send(MessageBuilder.create()
+ .setKey("b")
+ .setContent("B_2".getBytes()).build());
+ expected.put("b", "B_2".getBytes());
+
+ compactAndVerify(topic, expected);
+ }
+
+ @Test
+ public void testCompactedInOrder() throws Exception {
+ String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+ ProducerConfiguration producerConf = new ProducerConfiguration();
+ Producer producer = pulsarClient.createProducer(topic, producerConf);
+
+ producer.send(MessageBuilder.create()
+ .setKey("c")
+ .setContent("C_1".getBytes()).build());
+ producer.send(MessageBuilder.create()
+ .setKey("a")
+ .setContent("A_1".getBytes()).build());
+ producer.send(MessageBuilder.create()
+ .setKey("b")
+ .setContent("B_1".getBytes()).build());
+ producer.send(MessageBuilder.create()
+ .setKey("a")
+ .setContent("A_2".getBytes()).build());
+ Map<String, byte[]> expected = new HashMap<>();
+ expected.put("a", "A_2".getBytes());
+ expected.put("b", "B_1".getBytes());
+ expected.put("c", "C_1".getBytes());
+
+ List<String> keyOrder = compactAndVerify(topic, expected);
+
+ Assert.assertEquals(keyOrder, Lists.newArrayList("c", "b", "a"));
+ }
+
+ @Test(expectedExceptions = ExecutionException.class)
+ public void testCompactEmptyTopic() throws Exception {
+ String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+ // trigger creation of topic on server side
+ pulsarClient.subscribe(topic, "sub1").close();
+
+ BookKeeper bk = pulsar.getBookKeeperClientFactory().create(
+ this.conf, null);
+ Compactor compactor = new TwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+ compactor.compact(topic).get();
+ }
+
+ public ByteBuf extractPayload(RawMessage m) throws Exception {
+ ByteBuf payloadAndMetadata = m.getHeadersAndPayload();
+ Commands.readChecksum(payloadAndMetadata);
+ int metadataSize = payloadAndMetadata.readInt(); // metadata size
+ byte[] metadata = new byte[metadataSize];
+ payloadAndMetadata.readBytes(metadata);
+ return payloadAndMetadata.slice();
+ }
+}
--
To stop receiving notification emails like this one, please contact
mmerli@apache.org.