You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/10/25 21:36:22 UTC
[GitHub] [pulsar] heesung-sn opened a new pull request, #18195: [improve][broker] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
heesung-sn opened a new pull request, #18195:
URL: https://github.com/apache/pulsar/pull/18195
<!--
### Contribution Checklist
- PR title format should be *[type][component] summary*. For details, see *[Guideline - Pulsar PR Naming Convention](https://docs.google.com/document/d/1d8Pw6ZbWk-_pCKdOmdvx9rnhPiyuxwq60_TrD68d7BA/edit#heading=h.trs9rsex3xom)*.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
-->
<!-- or this PR is one task of an issue -->
Master Issue: https://github.com/apache/pulsar/issues/18099
### Motivation
<!-- Explain here the context, and why you're making that change. What is the problem you're trying to solve. -->
This PR supports `TopicCompactionStrategy` in `StrategicTwoPhaseCompactor` and `TableView` for the internal system topics, as proposed in PIP-215.
This PR does not expose StrategicTwoPhaseCompactor for customer topics yet. We want to expose this feature to users after proven to be stable on the new system topic in the new broker load balancer(in PIP-192) first.
### Modifications
This PR adds the following classes to implement `StrategicTwoPhaseCompactor.
- Added `StrategicTwoPhaseCompactor` that extends `TwoPhaseCompactor`.
- Added `TopicCompactionStrategy` interface.
- Added `TopicCompactionStrategy` logic in `StrategicTwoPhaseCompactor` and `TableViewImpl`.
- Added `CompactionReaderImpl` to scan topic messages in StrategicTwoPhaseCompactor. It cumulatively acknowledges the read messages at the end of StrategicTwoPhaseCompactor.
- Added `SubscriptionMode` and `SubscriptionInitialPosition` parameters in `ReaderConfigurationData` since the `CompactionReaderImpl` reader's subscription needs to be durable and needs to read from the earliest position.
- Added `RawBatchMessageContainerImpl` to batch and serialize messages in `StrategicTwoPhaseCompactor`.
This PR updates the `TableViewImpl` to use `TopicCompactionStrategy` in its data K,V map update logic.
- Added the `topicCompactionStrategy` member variable in `TableViewConfigurationData`
- Added the `listen()` interface in `TableView` to provide an option to call the listener actions only for the tail messages.
- Updated `TableViewImpl` and its `handleMessage()` function to consider `TopicCompactionStrategy` when updating the data K,V map.
This PR updated the compaction test classes to reuse the test cases.
This PR updated the modifiers of the parent classes of the added classes to access the member variables and member functions.
<!-- Describe the modifications you've done. -->
### Verifying this change
- [x] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
- Extended the compaction tests to cover the same logic passes with the `StrategicTwoPhaseCompactor`.
- Added other unit tests for the added classes.
### Does this pull request potentially affect one of the following parts:
*If the box was checked, please highlight the changes*
- [] Dependencies (add or upgrade a dependency)
- [x] The public API (table-view.listen() addition and reader and table-view configuration update)
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] Anything that affects deployment
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. Please attach the local preview screenshots (run `sh start.sh` at `pulsar/site2/website`) to your PR description, or else your PR might not get merged. -->
- [ ] `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 PR does not enable strategic compaction for the customer topics. We will update the doc when enabling this compaction for the customer topics.
### Matching PR in forked repository
PR in forked repository: https://github.com/heesung-sn/pulsar/pull/12
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
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 #18195: [improve][broker] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
Demogorgon314 commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1014971463
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.client.impl;
+
+import io.netty.buffer.ByteBuf;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.MessageCrypto;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
+import org.apache.pulsar.common.api.EncryptionContext;
+import org.apache.pulsar.common.api.proto.CompressionType;
+import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.compression.CompressionCodecNone;
+import org.apache.pulsar.common.compression.CompressionCodecProvider;
+import org.apache.pulsar.common.protocol.Commands;
+
+/**
+ * A raw batch message container without producer. (Used for StrategicTwoPhaseCompactor)
+ *
+ * incoming single messages:
+ * (k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)
+ *
+ * batched into single batch message:
+ * [(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;
+ public RawBatchMessageContainerImpl(int maxNumMessagesInBatch) {
+ super();
+ compressionType = CompressionType.NONE;
+ compressor = new CompressionCodecNone();
+ if (maxNumMessagesInBatch > 0) {
+ this.maxNumMessagesInBatch = maxNumMessagesInBatch;
+ }
+ }
+ private ByteBuf encrypt(ByteBuf compressedPayload) {
+ if (msgCrypto == null) {
+ return compressedPayload;
+ }
+ int maxSize = msgCrypto.getMaxOutputSize(compressedPayload.readableBytes());
+ ByteBuf encryptedPayload = allocator.buffer(maxSize);
+ ByteBuffer targetBuffer = encryptedPayload.nioBuffer(0, maxSize);
+
+ try {
+ msgCrypto.encrypt(encryptionKeys, cryptoKeyReader, () -> messageMetadata,
+ compressedPayload.nioBuffer(), targetBuffer);
+ } catch (PulsarClientException e) {
+ throw new RuntimeException("Failed to encrypt payload", e);
+ }
+ encryptedPayload.writerIndex(targetBuffer.remaining());
+ compressedPayload.release();
+ return encryptedPayload;
+ }
+
+ @Override
+ public ProducerImpl.OpSendMsg createOpSendMsg() {
+ throw new UnsupportedOperationException();
+ }
+ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
+ this.cryptoKeyReader = cryptoKeyReader;
+ }
+
+ public ByteBuf toByteBuf() {
+ if (numMessagesInBatch > 1) {
+ messageMetadata.setNumMessagesInBatch(numMessagesInBatch);
+ messageMetadata.setSequenceId(lowestSequenceId);
+ messageMetadata.setHighestSequenceId(highestSequenceId);
+ }
+ MessageImpl lastMessage = messages.get(messages.size() - 1);
+ MessageIdImpl lastMessageId = (MessageIdImpl) lastMessage.getMessageId();
+ MessageMetadata lastMessageMetadata = lastMessage.getMessageBuilder();
+
+ this.compressionType = lastMessageMetadata.getCompression();
+ this.compressor = CompressionCodecProvider.getCompressionCodec(lastMessageMetadata.getCompression());
+
+ if (!lastMessage.getEncryptionCtx().isEmpty()) {
+ EncryptionContext encryptionContext = (EncryptionContext) lastMessage.getEncryptionCtx().get();
+
+ if (cryptoKeyReader == null) {
+ throw new IllegalStateException("Messages are encrypted but no cryptoKeyReader is provided.");
Review Comment:
When the exception happens, we should call the `clear` method, 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] heesung-sn commented on a diff in pull request #18195: [improve][broker] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1015805463
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -0,0 +1,433 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.impl.BatchMessageIdImpl;
+import org.apache.pulsar.client.impl.CompactionReaderImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.RawBatchMessageContainerImpl;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects valid message(defined in the TopicCompactionStrategy.isValid())
+ * for each key in the topic. Then, the second pass writes these values
+ * to a ledger.
+ *
+ * <p>As the first pass caches the entire message(not just offset) for each key into a map,
+ * this compaction could be memory intensive if the message payload is large.
+ */
+public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
+ private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
+ private static final int MAX_OUTSTANDING = 500;
+ private final Duration phaseOneLoopReadTimeout;
+ private final RawBatchMessageContainerImpl batchMessageContainer;
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler,
+ int maxNumMessagesInBatch) {
+ super(conf, pulsar, bk, scheduler);
+ batchMessageContainer = new RawBatchMessageContainerImpl(maxNumMessagesInBatch);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+ }
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this(conf, pulsar, bk, scheduler, -1);
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy) {
+ return compact(topic, strategy, null);
+ }
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy,
+ CryptoKeyReader cryptoKeyReader) {
+ CompletableFuture<Consumer<T>> consumerFuture = new CompletableFuture<>();
+ if (cryptoKeyReader != null) {
+ batchMessageContainer.setCryptoKeyReader(cryptoKeyReader);
+ }
+ CompactionReaderImpl reader = CompactionReaderImpl.create(
+ (PulsarClientImpl) pulsar, strategy.getSchema(), topic, consumerFuture, cryptoKeyReader);
+
+ return consumerFuture.thenComposeAsync(__ -> compactAndCloseReader(reader, strategy), scheduler);
+ }
+
+ <T> CompletableFuture<Long> doCompaction(Reader<T> reader, TopicCompactionStrategy strategy) {
+
+ if (!(reader instanceof CompactionReaderImpl<T>)) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("reader has to be DelayedAckReaderImpl"));
+ }
+ return reader.hasMessageAvailableAsync()
+ .thenCompose(available -> {
+ if (available) {
+ return phaseOne(reader, strategy)
+ .thenCompose((result) -> phaseTwo(result, reader, bk));
+ } else {
+ log.info("Skip compaction of the empty topic {}", reader.getTopic());
+ return CompletableFuture.completedFuture(-1L);
+ }
+ });
+ }
+
+ <T> CompletableFuture<Long> compactAndCloseReader(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ mxBean.addCompactionStartOp(reader.getTopic());
+ doCompaction(reader, strategy).whenComplete(
+ (ledgerId, exception) -> {
+ log.info("Completed doCompaction ledgerId:{}", ledgerId);
+ reader.closeAsync().whenComplete((v, exception2) -> {
+ if (exception2 != null) {
+ log.warn("Error closing reader handle {}, ignoring", reader, exception2);
+ }
+ if (exception != null) {
+ // complete with original exception
+ mxBean.addCompactionEndOp(reader.getTopic(), false);
+ promise.completeExceptionally(exception);
+ } else {
+ mxBean.addCompactionEndOp(reader.getTopic(), true);
+ promise.complete(ledgerId);
+ }
+ });
+ });
+ return promise;
+ }
+
+ private <T> boolean doCompactMessage(
+ Message<T> msg, PhaseOneResult<T> result, TopicCompactionStrategy<T> strategy) {
+ Map<String, Message<T>> cache = result.cache;
+ String key = msg.getKey();
+
+ if (key == null) {
+ msg.release();
+ return true;
+ }
+ T val = msg.getValue();
+ Message<T> prev = cache.get(key);
+ T prevVal = prev == null ? null : prev.getValue();
+
+ if (!strategy.shouldKeepLeft(prevVal, val)) {
+ if (val != null && msg.size() > 0) {
+ cache.remove(key); // to reorder
+ cache.put(key, msg);
+ } else {
+ cache.remove(key);
+ msg.release();
+ }
+
+ if (prev != null) {
+ prev.release();
+ }
+
+ result.validCompactionCount.incrementAndGet();
+ return true;
+ } else {
+ msg.release();
+ result.invalidCompactionCount.incrementAndGet();
+ return false;
+ }
+
+ }
+
+ private static class PhaseOneResult<T> {
+ MessageId firstId;
+ MessageId lastId; // last read messageId
+ Map<String, Message<T>> cache;
+
+ AtomicInteger invalidCompactionCount;
+
+ AtomicInteger validCompactionCount;
+
+ AtomicInteger numReadMessages;
+
+ String topic;
+
+ PhaseOneResult(String topic) {
+ this.topic = topic;
+ cache = new LinkedHashMap<>();
+ invalidCompactionCount = new AtomicInteger();
+ validCompactionCount = new AtomicInteger();
+ numReadMessages = new AtomicInteger();
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "{Topic:%s, firstId:%s, lastId:%s, cache.size:%d, "
+ + "invalidCompactionCount:%d, validCompactionCount:%d, numReadMessages:%d}",
+ topic,
+ firstId != null ? firstId.toString() : "",
+ lastId != null ? lastId.toString() : "",
+ cache.size(),
+ invalidCompactionCount.get(),
+ validCompactionCount.get(),
+ numReadMessages.get());
+ }
+ }
+
+
+ private <T> CompletableFuture<PhaseOneResult> phaseOne(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<PhaseOneResult> promise = new CompletableFuture<>();
+ PhaseOneResult<T> result = new PhaseOneResult(reader.getTopic());
+
+ ((CompactionReaderImpl<T>) reader).getLastMessageIdAsync()
+ .thenAccept(lastMessageId -> {
+ log.info("Commencing phase one of compaction for {}, reading to {}",
+ reader.getTopic(), lastMessageId);
+ result.lastId = copyMessageId(lastMessageId);
+ phaseOneLoop(reader, promise, result, strategy);
+ }).exceptionally(ex -> {
+ promise.completeExceptionally(ex);
+ return null;
+ });
+
+ return promise;
+
+ }
+
+ private static MessageId copyMessageId(MessageId msgId) {
+ if (msgId instanceof BatchMessageIdImpl) {
+ BatchMessageIdImpl tempId = (BatchMessageIdImpl) msgId;
+ return new BatchMessageIdImpl(tempId);
+ } else if (msgId instanceof MessageIdImpl) {
+ MessageIdImpl tempId = (MessageIdImpl) msgId;
+ return new MessageIdImpl(tempId.getLedgerId(), tempId.getEntryId(),
+ tempId.getPartitionIndex());
+ } else {
+ throw new IllegalStateException("Unknown lastMessageId type");
+ }
+ }
+
+ private <T> void phaseOneLoop(Reader<T> reader, CompletableFuture<PhaseOneResult> promise,
+ PhaseOneResult<T> result, TopicCompactionStrategy<T> strategy) {
+
+ if (promise.isDone()) {
+ return;
+ }
+
+ CompletableFuture<Message<T>> future = reader.readNextAsync();
+ FutureUtil.addTimeoutHandling(future,
+ phaseOneLoopReadTimeout, scheduler,
+ () -> FutureUtil.createTimeoutException("Timeout", getClass(),
+ "phaseOneLoop(...)"));
+
+ future.thenAcceptAsync(msg -> {
+
+ MessageId id = msg.getMessageId();
+ boolean completed = false;
+ if (result.lastId.compareTo(id) == 0) {
+ completed = true;
+ }
+
+ result.numReadMessages.incrementAndGet();
+ mxBean.addCompactionReadOp(reader.getTopic(), msg.size());
+ if (doCompactMessage(msg, result, strategy)) {
+ mxBean.addCompactionRemovedEvent(reader.getTopic());
+ }
+ //set ids in the result
+ if (result.firstId == null) {
+ result.firstId = copyMessageId(id);
+ log.info("Resetting cursor to firstId:{}", result.firstId);
+ try {
+ reader.seek(result.firstId);
+ } catch (PulsarClientException e) {
+ throw new RuntimeException("Failed to reset the cursor to firstId:" + result.firstId, e);
+ }
+ }
+ if (completed) {
+ promise.complete(result);
+ } else {
+ phaseOneLoop(reader, promise, result, strategy);
+ }
+
+ }, scheduler).exceptionally(ex -> {
+ promise.completeExceptionally(ex);
+ return null;
+ });
+
+ }
+
+ private <T> CompletableFuture<Long> phaseTwo(PhaseOneResult<T> phaseOneResult, Reader<T> reader, BookKeeper bk) {
+ log.info("Completed phase one. Result:{}. ", phaseOneResult);
+ Map<String, byte[]> metadata =
+ LedgerMetadataUtils.buildMetadataForCompactedLedger(
+ phaseOneResult.topic, phaseOneResult.lastId.toByteArray());
+ return createLedger(bk, metadata)
+ .thenCompose((ledger) -> {
+ log.info(
+ "Commencing phase two of compaction for {}, from {} to {}, compacting {} keys to ledger {}",
+ phaseOneResult.topic, phaseOneResult.firstId, phaseOneResult.lastId,
+ phaseOneResult.cache.size(), ledger.getId());
+ return runPhaseTwo(phaseOneResult, reader, ledger, bk);
+ });
+ }
+
+ private <T> CompletableFuture<Long> runPhaseTwo(
+ PhaseOneResult<T> phaseOneResult, Reader<T> reader, LedgerHandle ledger, BookKeeper bk) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ Semaphore outstanding = new Semaphore(MAX_OUTSTANDING);
+ CompletableFuture<Void> loopPromise = new CompletableFuture<>();
+ phaseTwoLoop(phaseOneResult.topic, phaseOneResult.cache.values().iterator(), ledger,
+ outstanding, loopPromise);
+ loopPromise.thenCompose((v) -> {
+ log.info("Flushing batch container numMessagesInBatch:{}",
+ batchMessageContainer.getNumMessagesInBatch());
+ return addToCompactedLedger(ledger, null, reader.getTopic(), outstanding)
+ .whenComplete((res, exception2) -> {
+ if (exception2 != null) {
+ promise.completeExceptionally(exception2);
+ return;
+ }
+ });
+ })
+ .thenCompose(v -> {
+ log.info("Acking ledger id {}", phaseOneResult.firstId);
+ return ((CompactionReaderImpl<T>) reader)
+ .acknowledgeCumulativeAsync(
+ phaseOneResult.lastId, Map.of(COMPACTED_TOPIC_LEDGER_PROPERTY,
+ ledger.getId()));
+ })
+ .thenCompose((v) -> closeLedger(ledger))
+ .whenComplete((v, exception) -> {
+ if (exception != null) {
+ deleteLedger(bk, ledger).whenComplete((res2, exception2) -> {
+ if (exception2 != null) {
+ log.error("Cleanup of ledger {} for failed", ledger, exception2);
+ }
+ // complete with original exception
+ promise.completeExceptionally(exception);
+ });
+ } else {
+ log.info("kept ledger:{}", ledger.getId());
+ promise.complete(ledger.getId());
+ }
+ });
+
+ return promise;
+ }
+
+ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
+ LedgerHandle lh, Semaphore outstanding,
+ CompletableFuture<Void> promise) {
+ if (promise.isDone()) {
+ return;
+ }
+ CompletableFuture.runAsync(() -> {
+ if (reader.hasNext()) {
+ Message<T> message = reader.next();
+ mxBean.addCompactionReadOp(topic, message.size());
+ addToCompactedLedger(lh, message, topic, outstanding)
+ .whenComplete((res, exception2) -> {
+ outstanding.release();
+ if (exception2 != null) {
+ promise.completeExceptionally(exception2);
+ return;
+ }
+ });
+ phaseTwoLoop(topic, reader, lh, outstanding, promise);
Review Comment:
No. We don't want to wait for each `addToCompactedLedger` call to complete for high concurrency.
--
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 #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1028717158
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.client.impl;
+
+import io.netty.buffer.ByteBuf;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.MessageCrypto;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
+import org.apache.pulsar.common.api.EncryptionContext;
+import org.apache.pulsar.common.api.proto.CompressionType;
+import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.compression.CompressionCodecNone;
+import org.apache.pulsar.common.compression.CompressionCodecProvider;
+import org.apache.pulsar.common.protocol.Commands;
+
+/**
+ * A raw batch message container without producer. (Used for StrategicTwoPhaseCompactor)
+ *
+ * incoming single messages:
+ * (k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)
+ *
+ * batched into single batch message:
+ * [(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]
+ */
+public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
Review Comment:
But I am not fully convinced this mock producer is better than the null check. Can you explain why this could be better? The base class sets producer later too, it also has potential that producer could be null.
Obviously, We dont want to create ledger entry per message. especially this message payload is very small.
--
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 #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1037891433
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -0,0 +1,434 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.impl.BatchMessageIdImpl;
+import org.apache.pulsar.client.impl.CompactionReaderImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.RawBatchMessageContainerImpl;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects valid message(defined in the TopicCompactionStrategy.isValid())
+ * for each key in the topic. Then, the second pass writes these values
+ * to a ledger.
+ *
+ * <p>As the first pass caches the entire message(not just offset) for each key into a map,
+ * this compaction could be memory intensive if the message payload is large.
+ */
+public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
+ private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
+ private static final int MAX_OUTSTANDING = 500;
+ private final Duration phaseOneLoopReadTimeout;
+ private final RawBatchMessageContainerImpl batchMessageContainer;
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler,
+ int maxNumMessagesInBatch) {
+ super(conf, pulsar, bk, scheduler);
+ batchMessageContainer = new RawBatchMessageContainerImpl(maxNumMessagesInBatch);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+ }
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this(conf, pulsar, bk, scheduler, -1);
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy) {
+ return compact(topic, strategy, null);
+ }
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy,
+ CryptoKeyReader cryptoKeyReader) {
+ CompletableFuture<Consumer<T>> consumerFuture = new CompletableFuture<>();
+ if (cryptoKeyReader != null) {
+ batchMessageContainer.setCryptoKeyReader(cryptoKeyReader);
+ }
+ CompactionReaderImpl reader = CompactionReaderImpl.create(
+ (PulsarClientImpl) pulsar, strategy.getSchema(), topic, consumerFuture, cryptoKeyReader);
+
+ return consumerFuture.thenComposeAsync(__ -> compactAndCloseReader(reader, strategy), scheduler);
+ }
+
+ <T> CompletableFuture<Long> doCompaction(Reader<T> reader, TopicCompactionStrategy strategy) {
+
+ if (!(reader instanceof CompactionReaderImpl<T>)) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("reader has to be DelayedAckReaderImpl"));
+ }
+ return reader.hasMessageAvailableAsync()
Review Comment:
@codelipenghui
AFAIK, this deserialization is inevitable( both in the prefilter or postfilter method), do you have other suggestions?
--
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 #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1033143208
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewConfigurationData.java:
##########
@@ -29,13 +29,15 @@ public class TableViewConfigurationData implements Serializable, Cloneable {
private String topicName = null;
private long autoUpdatePartitionsSeconds = 60;
+ private String topicCompactionStrategy;
Review Comment:
> And why not use TopicCompactionStrategy topicCompactionStrategy directly? If we don't want to expose it as a public API, we can just ignore the serialization of the topicCompactionStrategy
I couldn't find a strong benefit to pass 'TopicCompactionStrategy' obj in this config.
I thought this config needs to be light-weight 'configs' instead of passing heavy objects.
--
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 #18195: [improve][broker] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
Demogorgon314 commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1016625534
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -0,0 +1,433 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.impl.BatchMessageIdImpl;
+import org.apache.pulsar.client.impl.CompactionReaderImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.RawBatchMessageContainerImpl;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects valid message(defined in the TopicCompactionStrategy.isValid())
+ * for each key in the topic. Then, the second pass writes these values
+ * to a ledger.
+ *
+ * <p>As the first pass caches the entire message(not just offset) for each key into a map,
+ * this compaction could be memory intensive if the message payload is large.
+ */
+public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
+ private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
+ private static final int MAX_OUTSTANDING = 500;
+ private final Duration phaseOneLoopReadTimeout;
+ private final RawBatchMessageContainerImpl batchMessageContainer;
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler,
+ int maxNumMessagesInBatch) {
+ super(conf, pulsar, bk, scheduler);
+ batchMessageContainer = new RawBatchMessageContainerImpl(maxNumMessagesInBatch);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+ }
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this(conf, pulsar, bk, scheduler, -1);
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy) {
+ return compact(topic, strategy, null);
+ }
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy,
+ CryptoKeyReader cryptoKeyReader) {
+ CompletableFuture<Consumer<T>> consumerFuture = new CompletableFuture<>();
+ if (cryptoKeyReader != null) {
+ batchMessageContainer.setCryptoKeyReader(cryptoKeyReader);
+ }
+ CompactionReaderImpl reader = CompactionReaderImpl.create(
+ (PulsarClientImpl) pulsar, strategy.getSchema(), topic, consumerFuture, cryptoKeyReader);
+
+ return consumerFuture.thenComposeAsync(__ -> compactAndCloseReader(reader, strategy), scheduler);
+ }
+
+ <T> CompletableFuture<Long> doCompaction(Reader<T> reader, TopicCompactionStrategy strategy) {
+
+ if (!(reader instanceof CompactionReaderImpl<T>)) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("reader has to be DelayedAckReaderImpl"));
+ }
+ return reader.hasMessageAvailableAsync()
+ .thenCompose(available -> {
+ if (available) {
+ return phaseOne(reader, strategy)
+ .thenCompose((result) -> phaseTwo(result, reader, bk));
+ } else {
+ log.info("Skip compaction of the empty topic {}", reader.getTopic());
+ return CompletableFuture.completedFuture(-1L);
+ }
+ });
+ }
+
+ <T> CompletableFuture<Long> compactAndCloseReader(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ mxBean.addCompactionStartOp(reader.getTopic());
+ doCompaction(reader, strategy).whenComplete(
+ (ledgerId, exception) -> {
+ log.info("Completed doCompaction ledgerId:{}", ledgerId);
+ reader.closeAsync().whenComplete((v, exception2) -> {
+ if (exception2 != null) {
+ log.warn("Error closing reader handle {}, ignoring", reader, exception2);
+ }
+ if (exception != null) {
+ // complete with original exception
+ mxBean.addCompactionEndOp(reader.getTopic(), false);
+ promise.completeExceptionally(exception);
+ } else {
+ mxBean.addCompactionEndOp(reader.getTopic(), true);
+ promise.complete(ledgerId);
+ }
+ });
+ });
+ return promise;
+ }
+
+ private <T> boolean doCompactMessage(
+ Message<T> msg, PhaseOneResult<T> result, TopicCompactionStrategy<T> strategy) {
+ Map<String, Message<T>> cache = result.cache;
+ String key = msg.getKey();
+
+ if (key == null) {
+ msg.release();
+ return true;
+ }
+ T val = msg.getValue();
+ Message<T> prev = cache.get(key);
+ T prevVal = prev == null ? null : prev.getValue();
+
+ if (!strategy.shouldKeepLeft(prevVal, val)) {
+ if (val != null && msg.size() > 0) {
+ cache.remove(key); // to reorder
+ cache.put(key, msg);
+ } else {
+ cache.remove(key);
+ msg.release();
+ }
+
+ if (prev != null) {
+ prev.release();
+ }
+
+ result.validCompactionCount.incrementAndGet();
+ return true;
+ } else {
+ msg.release();
+ result.invalidCompactionCount.incrementAndGet();
+ return false;
+ }
+
+ }
+
+ private static class PhaseOneResult<T> {
+ MessageId firstId;
+ MessageId lastId; // last read messageId
+ Map<String, Message<T>> cache;
+
+ AtomicInteger invalidCompactionCount;
+
+ AtomicInteger validCompactionCount;
+
+ AtomicInteger numReadMessages;
+
+ String topic;
+
+ PhaseOneResult(String topic) {
+ this.topic = topic;
+ cache = new LinkedHashMap<>();
+ invalidCompactionCount = new AtomicInteger();
+ validCompactionCount = new AtomicInteger();
+ numReadMessages = new AtomicInteger();
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "{Topic:%s, firstId:%s, lastId:%s, cache.size:%d, "
+ + "invalidCompactionCount:%d, validCompactionCount:%d, numReadMessages:%d}",
+ topic,
+ firstId != null ? firstId.toString() : "",
+ lastId != null ? lastId.toString() : "",
+ cache.size(),
+ invalidCompactionCount.get(),
+ validCompactionCount.get(),
+ numReadMessages.get());
+ }
+ }
+
+
+ private <T> CompletableFuture<PhaseOneResult> phaseOne(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<PhaseOneResult> promise = new CompletableFuture<>();
+ PhaseOneResult<T> result = new PhaseOneResult(reader.getTopic());
+
+ ((CompactionReaderImpl<T>) reader).getLastMessageIdAsync()
+ .thenAccept(lastMessageId -> {
+ log.info("Commencing phase one of compaction for {}, reading to {}",
+ reader.getTopic(), lastMessageId);
+ result.lastId = copyMessageId(lastMessageId);
+ phaseOneLoop(reader, promise, result, strategy);
+ }).exceptionally(ex -> {
+ promise.completeExceptionally(ex);
+ return null;
+ });
+
+ return promise;
+
+ }
+
+ private static MessageId copyMessageId(MessageId msgId) {
+ if (msgId instanceof BatchMessageIdImpl) {
+ BatchMessageIdImpl tempId = (BatchMessageIdImpl) msgId;
+ return new BatchMessageIdImpl(tempId);
+ } else if (msgId instanceof MessageIdImpl) {
+ MessageIdImpl tempId = (MessageIdImpl) msgId;
+ return new MessageIdImpl(tempId.getLedgerId(), tempId.getEntryId(),
+ tempId.getPartitionIndex());
+ } else {
+ throw new IllegalStateException("Unknown lastMessageId type");
+ }
+ }
+
+ private <T> void phaseOneLoop(Reader<T> reader, CompletableFuture<PhaseOneResult> promise,
+ PhaseOneResult<T> result, TopicCompactionStrategy<T> strategy) {
+
+ if (promise.isDone()) {
+ return;
+ }
+
+ CompletableFuture<Message<T>> future = reader.readNextAsync();
+ FutureUtil.addTimeoutHandling(future,
+ phaseOneLoopReadTimeout, scheduler,
+ () -> FutureUtil.createTimeoutException("Timeout", getClass(),
+ "phaseOneLoop(...)"));
+
+ future.thenAcceptAsync(msg -> {
+
+ MessageId id = msg.getMessageId();
+ boolean completed = false;
+ if (result.lastId.compareTo(id) == 0) {
+ completed = true;
+ }
+
+ result.numReadMessages.incrementAndGet();
+ mxBean.addCompactionReadOp(reader.getTopic(), msg.size());
+ if (doCompactMessage(msg, result, strategy)) {
+ mxBean.addCompactionRemovedEvent(reader.getTopic());
+ }
+ //set ids in the result
+ if (result.firstId == null) {
+ result.firstId = copyMessageId(id);
+ log.info("Resetting cursor to firstId:{}", result.firstId);
+ try {
+ reader.seek(result.firstId);
+ } catch (PulsarClientException e) {
+ throw new RuntimeException("Failed to reset the cursor to firstId:" + result.firstId, e);
+ }
+ }
+ if (completed) {
+ promise.complete(result);
+ } else {
+ phaseOneLoop(reader, promise, result, strategy);
+ }
+
+ }, scheduler).exceptionally(ex -> {
+ promise.completeExceptionally(ex);
+ return null;
+ });
+
+ }
+
+ private <T> CompletableFuture<Long> phaseTwo(PhaseOneResult<T> phaseOneResult, Reader<T> reader, BookKeeper bk) {
+ log.info("Completed phase one. Result:{}. ", phaseOneResult);
+ Map<String, byte[]> metadata =
+ LedgerMetadataUtils.buildMetadataForCompactedLedger(
+ phaseOneResult.topic, phaseOneResult.lastId.toByteArray());
+ return createLedger(bk, metadata)
+ .thenCompose((ledger) -> {
+ log.info(
+ "Commencing phase two of compaction for {}, from {} to {}, compacting {} keys to ledger {}",
+ phaseOneResult.topic, phaseOneResult.firstId, phaseOneResult.lastId,
+ phaseOneResult.cache.size(), ledger.getId());
+ return runPhaseTwo(phaseOneResult, reader, ledger, bk);
+ });
+ }
+
+ private <T> CompletableFuture<Long> runPhaseTwo(
+ PhaseOneResult<T> phaseOneResult, Reader<T> reader, LedgerHandle ledger, BookKeeper bk) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ Semaphore outstanding = new Semaphore(MAX_OUTSTANDING);
+ CompletableFuture<Void> loopPromise = new CompletableFuture<>();
+ phaseTwoLoop(phaseOneResult.topic, phaseOneResult.cache.values().iterator(), ledger,
+ outstanding, loopPromise);
+ loopPromise.thenCompose((v) -> {
+ log.info("Flushing batch container numMessagesInBatch:{}",
+ batchMessageContainer.getNumMessagesInBatch());
+ return addToCompactedLedger(ledger, null, reader.getTopic(), outstanding)
+ .whenComplete((res, exception2) -> {
+ if (exception2 != null) {
+ promise.completeExceptionally(exception2);
+ return;
+ }
+ });
+ })
+ .thenCompose(v -> {
+ log.info("Acking ledger id {}", phaseOneResult.firstId);
+ return ((CompactionReaderImpl<T>) reader)
+ .acknowledgeCumulativeAsync(
+ phaseOneResult.lastId, Map.of(COMPACTED_TOPIC_LEDGER_PROPERTY,
+ ledger.getId()));
+ })
+ .thenCompose((v) -> closeLedger(ledger))
+ .whenComplete((v, exception) -> {
+ if (exception != null) {
+ deleteLedger(bk, ledger).whenComplete((res2, exception2) -> {
+ if (exception2 != null) {
+ log.error("Cleanup of ledger {} for failed", ledger, exception2);
+ }
+ // complete with original exception
+ promise.completeExceptionally(exception);
+ });
+ } else {
+ log.info("kept ledger:{}", ledger.getId());
+ promise.complete(ledger.getId());
+ }
+ });
+
+ return promise;
+ }
+
+ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
+ LedgerHandle lh, Semaphore outstanding,
+ CompletableFuture<Void> promise) {
+ if (promise.isDone()) {
+ return;
+ }
+ CompletableFuture.runAsync(() -> {
+ if (reader.hasNext()) {
+ Message<T> message = reader.next();
+ mxBean.addCompactionReadOp(topic, message.size());
+ addToCompactedLedger(lh, message, topic, outstanding)
+ .whenComplete((res, exception2) -> {
+ outstanding.release();
+ if (exception2 != null) {
+ promise.completeExceptionally(exception2);
+ return;
+ }
+ });
+ phaseTwoLoop(topic, reader, lh, outstanding, promise);
Review Comment:
I see.
--
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 #18195: [improve][broker] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
Demogorgon314 commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1012597347
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java:
##########
@@ -51,13 +52,15 @@
private final List<BiConsumer<String, T>> listeners;
private final ReentrantLock listenersMutex;
+ private TopicCompactionStrategy<T> compactionStrategy;
TableViewImpl(PulsarClientImpl client, Schema<T> schema, TableViewConfigurationData conf) {
this.conf = conf;
this.data = new ConcurrentHashMap<>();
this.immutableData = Collections.unmodifiableMap(data);
this.listeners = new ArrayList<>();
this.listenersMutex = new ReentrantLock();
+ this.compactionStrategy = TopicCompactionStrategy.load(conf.getTopicCompactionStrategy());
Review Comment:
Do we have a unit test to cover this new feature?
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java:
##########
@@ -164,8 +164,8 @@ public Message<T> readNext() throws PulsarClientException {
@Override
public Message<T> readNext(int timeout, TimeUnit unit) throws PulsarClientException {
- Message<T> msg = consumer.receive(timeout, unit);
+ Message<T> msg = consumer.receive(timeout, unit);
Review Comment:
Please remove the useless change.
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -0,0 +1,433 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.impl.BatchMessageIdImpl;
+import org.apache.pulsar.client.impl.CompactionReaderImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.RawBatchMessageContainerImpl;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects valid message(defined in the TopicCompactionStrategy.isValid())
+ * for each key in the topic. Then, the second pass writes these values
+ * to a ledger.
+ *
+ * <p>As the first pass caches the entire message(not just offset) for each key into a map,
+ * this compaction could be memory intensive if the message payload is large.
+ */
+public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
+ private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
+ private static final int MAX_OUTSTANDING = 500;
+ private final Duration phaseOneLoopReadTimeout;
+ private final RawBatchMessageContainerImpl batchMessageContainer;
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler,
+ int maxNumMessagesInBatch) {
+ super(conf, pulsar, bk, scheduler);
+ batchMessageContainer = new RawBatchMessageContainerImpl(maxNumMessagesInBatch);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+ }
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this(conf, pulsar, bk, scheduler, -1);
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy) {
+ return compact(topic, strategy, null);
+ }
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy,
+ CryptoKeyReader cryptoKeyReader) {
+ CompletableFuture<Consumer<T>> consumerFuture = new CompletableFuture<>();
+ if (cryptoKeyReader != null) {
+ batchMessageContainer.setCryptoKeyReader(cryptoKeyReader);
+ }
+ CompactionReaderImpl reader = CompactionReaderImpl.create(
+ (PulsarClientImpl) pulsar, strategy.getSchema(), topic, consumerFuture, cryptoKeyReader);
+
+ return consumerFuture.thenComposeAsync(__ -> compactAndCloseReader(reader, strategy), scheduler);
+ }
+
+ <T> CompletableFuture<Long> doCompaction(Reader<T> reader, TopicCompactionStrategy strategy) {
+
+ if (!(reader instanceof CompactionReaderImpl<T>)) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("reader has to be DelayedAckReaderImpl"));
+ }
+ return reader.hasMessageAvailableAsync()
+ .thenCompose(available -> {
+ if (available) {
+ return phaseOne(reader, strategy)
+ .thenCompose((result) -> phaseTwo(result, reader, bk));
+ } else {
+ log.info("Skip compaction of the empty topic {}", reader.getTopic());
+ return CompletableFuture.completedFuture(-1L);
+ }
+ });
+ }
+
+ <T> CompletableFuture<Long> compactAndCloseReader(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ mxBean.addCompactionStartOp(reader.getTopic());
+ doCompaction(reader, strategy).whenComplete(
+ (ledgerId, exception) -> {
+ log.info("Completed doCompaction ledgerId:{}", ledgerId);
+ reader.closeAsync().whenComplete((v, exception2) -> {
+ if (exception2 != null) {
+ log.warn("Error closing reader handle {}, ignoring", reader, exception2);
+ }
+ if (exception != null) {
+ // complete with original exception
+ mxBean.addCompactionEndOp(reader.getTopic(), false);
+ promise.completeExceptionally(exception);
+ } else {
+ mxBean.addCompactionEndOp(reader.getTopic(), true);
+ promise.complete(ledgerId);
+ }
+ });
+ });
+ return promise;
+ }
+
+ private <T> boolean doCompactMessage(Message<T> msg, PhaseOneResult<T> result, TopicCompactionStrategy strategy) {
+ Map<String, Message<T>> cache = result.cache;
+ String key = msg.getKey();
+
+ if (key == null) {
+ msg.release();
+ return true;
+ }
+ T val = msg.getValue();
+ Message<T> prev = cache.get(key);
+ T prevVal = prev == null ? null : prev.getValue();
+
+ if (strategy.isValid(prevVal, val)) {
+ if (val != null && msg.size() > 0) {
+ cache.remove(key); // to reorder
+ cache.put(key, msg);
+ } else {
+ cache.remove(key);
+ msg.release();
+ }
+
+ if (prev != null) {
+ prev.release();
+ }
+
+ result.validCompactionCount.incrementAndGet();
+ return true;
+ } else {
+ msg.release();
+ result.invalidCompactionCount.incrementAndGet();
+ return false;
+ }
+
+ }
+
+ private static class PhaseOneResult<T> {
+ MessageId firstId;
+ //MessageId to; // last undeleted messageId
+ MessageId lastId; // last read messageId
+ Map<String, Message<T>> cache;
+
+ AtomicInteger invalidCompactionCount;
+
+ AtomicInteger validCompactionCount;
+
+ AtomicInteger numReadMessages;
+
+ String topic;
+
+ PhaseOneResult(String topic) {
+ this.topic = topic;
+ cache = new LinkedHashMap<>();
+ invalidCompactionCount = new AtomicInteger();
+ validCompactionCount = new AtomicInteger();
+ numReadMessages = new AtomicInteger();
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "{Topic:%s, firstId:%s, lastId:%s, cache.size:%d, "
+ + "invalidCompactionCount:%d, validCompactionCount:%d, numReadMessages:%d}",
+ topic,
+ firstId != null ? firstId.toString() : "",
+ lastId != null ? lastId.toString() : "",
+ cache.size(),
+ invalidCompactionCount.get(),
+ validCompactionCount.get(),
+ numReadMessages.get());
+ }
+ }
+
+
+ private <T> CompletableFuture<PhaseOneResult> phaseOne(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<PhaseOneResult> promise = new CompletableFuture<>();
+ PhaseOneResult<T> result = new PhaseOneResult(reader.getTopic());
+
+ ((CompactionReaderImpl<T>) reader).getLastMessageIdAsync()
+ .thenAccept(lastMessageId -> {
+ log.info("Commencing phase one of compaction for {}, reading to {}",
+ reader.getTopic(), lastMessageId);
+ result.lastId = copyMessageId(lastMessageId);
+ phaseOneLoop(reader, promise, result, strategy);
+ }).exceptionally(ex -> {
+ promise.completeExceptionally(ex);
+ return null;
+ });
+
+ return promise;
+
+ }
+
+ private static MessageId copyMessageId(MessageId msgId) {
+ if (msgId instanceof BatchMessageIdImpl) {
+ BatchMessageIdImpl tempId = (BatchMessageIdImpl) msgId;
+ return new BatchMessageIdImpl(tempId);
+ } else if (msgId instanceof MessageIdImpl) {
+ MessageIdImpl tempId = (MessageIdImpl) msgId;
+ return new MessageIdImpl(tempId.getLedgerId(), tempId.getEntryId(),
+ tempId.getPartitionIndex());
+ } else {
+ throw new IllegalStateException("Unknown lastMessageId type");
+ }
+ }
+
+ private <T> void phaseOneLoop(Reader<T> reader, CompletableFuture<PhaseOneResult> promise,
+ PhaseOneResult<T> result, TopicCompactionStrategy strategy) {
Review Comment:
```suggestion
private <T> void phaseOneLoop(Reader<T> reader, CompletableFuture<PhaseOneResult> promise,
PhaseOneResult<T> result, TopicCompactionStrategy<T> strategy) {
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -0,0 +1,433 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.impl.BatchMessageIdImpl;
+import org.apache.pulsar.client.impl.CompactionReaderImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.RawBatchMessageContainerImpl;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects valid message(defined in the TopicCompactionStrategy.isValid())
+ * for each key in the topic. Then, the second pass writes these values
+ * to a ledger.
+ *
+ * <p>As the first pass caches the entire message(not just offset) for each key into a map,
+ * this compaction could be memory intensive if the message payload is large.
+ */
+public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
+ private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
+ private static final int MAX_OUTSTANDING = 500;
+ private final Duration phaseOneLoopReadTimeout;
+ private final RawBatchMessageContainerImpl batchMessageContainer;
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler,
+ int maxNumMessagesInBatch) {
+ super(conf, pulsar, bk, scheduler);
+ batchMessageContainer = new RawBatchMessageContainerImpl(maxNumMessagesInBatch);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+ }
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this(conf, pulsar, bk, scheduler, -1);
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy) {
+ return compact(topic, strategy, null);
+ }
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy,
+ CryptoKeyReader cryptoKeyReader) {
+ CompletableFuture<Consumer<T>> consumerFuture = new CompletableFuture<>();
+ if (cryptoKeyReader != null) {
+ batchMessageContainer.setCryptoKeyReader(cryptoKeyReader);
+ }
+ CompactionReaderImpl reader = CompactionReaderImpl.create(
+ (PulsarClientImpl) pulsar, strategy.getSchema(), topic, consumerFuture, cryptoKeyReader);
+
+ return consumerFuture.thenComposeAsync(__ -> compactAndCloseReader(reader, strategy), scheduler);
+ }
+
+ <T> CompletableFuture<Long> doCompaction(Reader<T> reader, TopicCompactionStrategy strategy) {
+
+ if (!(reader instanceof CompactionReaderImpl<T>)) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("reader has to be DelayedAckReaderImpl"));
+ }
+ return reader.hasMessageAvailableAsync()
+ .thenCompose(available -> {
+ if (available) {
+ return phaseOne(reader, strategy)
+ .thenCompose((result) -> phaseTwo(result, reader, bk));
+ } else {
+ log.info("Skip compaction of the empty topic {}", reader.getTopic());
+ return CompletableFuture.completedFuture(-1L);
+ }
+ });
+ }
+
+ <T> CompletableFuture<Long> compactAndCloseReader(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ mxBean.addCompactionStartOp(reader.getTopic());
+ doCompaction(reader, strategy).whenComplete(
+ (ledgerId, exception) -> {
+ log.info("Completed doCompaction ledgerId:{}", ledgerId);
+ reader.closeAsync().whenComplete((v, exception2) -> {
+ if (exception2 != null) {
+ log.warn("Error closing reader handle {}, ignoring", reader, exception2);
+ }
+ if (exception != null) {
+ // complete with original exception
+ mxBean.addCompactionEndOp(reader.getTopic(), false);
+ promise.completeExceptionally(exception);
+ } else {
+ mxBean.addCompactionEndOp(reader.getTopic(), true);
+ promise.complete(ledgerId);
+ }
+ });
+ });
+ return promise;
+ }
+
+ private <T> boolean doCompactMessage(Message<T> msg, PhaseOneResult<T> result, TopicCompactionStrategy strategy) {
Review Comment:
```suggestion
private <T> boolean doCompactMessage(Message<T> msg, PhaseOneResult<T> result, TopicCompactionStrategy<T> strategy) {
```
##########
pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicCompactionStrategy.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * 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.common.topics;
+
+import org.apache.pulsar.client.api.Schema;
+
+/**
+ * Defines a strategy to compact messages in a topic.
+ * This strategy can be passed to Topic Compactor and Table View to select messages in a specific way.
+ *
+ * Examples:
+ *
+ * TopicCompactionStrategy strategy = new MyTopicCompactionStrategy();
+ *
+ * // Run topic compaction by the compaction strategy.
+ * // While compacting messages for each key,
+ * // it will choose messages only if TopicCompactionStrategy.valid(prev, cur) returns true.
+ * StrategicTwoPhaseCompactor compactor = new StrategicTwoPhaseCompactor(...);
+ * compactor.compact(topic, strategy);
+ *
+ * // Run table view by the compaction strategy.
+ * // While updating messages in the table view <key,value> map,
+ * // it will choose messages only if TopicCompactionStrategy.valid(prev, cur) returns true.
+ * TableView tableView = pulsar.getClient().newTableViewBuilder(strategy.getSchema())
+ * .topic(topic)
+ * .loadConf(Map.of(
+ * "topicCompactionStrategy", strategy.getClass().getCanonicalName()))
+ * .create();
+ */
+public interface TopicCompactionStrategy<T> {
+
+ /**
+ * Returns the schema object for this strategy.
+ * @return
+ */
+ Schema<T> getSchema();
+ /**
+ * Tests if the current message is valid compared to the previous message for the same key.
+ *
+ * @param prev previous message
+ * @param cur current message
+ * @return True if the prev to the cur message transition is valid. Otherwise, false.
+ */
+ boolean isValid(T prev, T cur);
+
+
+ static TopicCompactionStrategy load(String topicCompactionStrategy) {
Review Comment:
```suggestion
static <T> TopicCompactionStrategy<T> load(String topicCompactionStrategy) {
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/CompactionReaderImpl.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.client.impl;
+
+import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.api.proto.CommandAck;
+
+/**
+ * An extended ReaderImpl used for StrategicTwoPhaseCompactor.
+ * The compaction consumer subscription is durable and consumes compacted messages from the earliest position.
+ * It does not acknowledge the message after each read. (needs to call acknowledgeCumulativeAsync to ack messages.)
+ */
+@Slf4j
+public class CompactionReaderImpl<T> extends ReaderImpl<T> {
+
+ ConsumerBase<T> consumer;
+
+ ReaderConfigurationData<T> readerConfiguration;
+ private CompactionReaderImpl(PulsarClientImpl client, ReaderConfigurationData<T> readerConfiguration,
+ ExecutorProvider executorProvider, CompletableFuture<Consumer<T>> consumerFuture,
+ Schema<T> schema) {
+ super(client, readerConfiguration, executorProvider, consumerFuture, schema);
+ this.readerConfiguration = readerConfiguration;
+ this.consumer = getConsumer();
+ }
+
+ public static <T> CompactionReaderImpl<T> create(PulsarClientImpl client, Schema<T> schema, String topic,
+ CompletableFuture<Consumer<T>> consumerFuture,
+ CryptoKeyReader cryptoKeyReader) {
+ ReaderConfigurationData conf = new ReaderConfigurationData<>();
Review Comment:
Add generics.
```suggestion
ReaderConfigurationData<T> conf = new ReaderConfigurationData<>();
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/CompactionReaderImpl.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.client.impl;
+
+import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.api.proto.CommandAck;
+
+/**
+ * An extended ReaderImpl used for StrategicTwoPhaseCompactor.
+ * The compaction consumer subscription is durable and consumes compacted messages from the earliest position.
+ * It does not acknowledge the message after each read. (needs to call acknowledgeCumulativeAsync to ack messages.)
+ */
+@Slf4j
+public class CompactionReaderImpl<T> extends ReaderImpl<T> {
+
+ ConsumerBase<T> consumer;
+
+ ReaderConfigurationData<T> readerConfiguration;
+ private CompactionReaderImpl(PulsarClientImpl client, ReaderConfigurationData<T> readerConfiguration,
+ ExecutorProvider executorProvider, CompletableFuture<Consumer<T>> consumerFuture,
+ Schema<T> schema) {
+ super(client, readerConfiguration, executorProvider, consumerFuture, schema);
+ this.readerConfiguration = readerConfiguration;
+ this.consumer = getConsumer();
+ }
+
+ public static <T> CompactionReaderImpl<T> create(PulsarClientImpl client, Schema<T> schema, String topic,
+ CompletableFuture<Consumer<T>> consumerFuture,
+ CryptoKeyReader cryptoKeyReader) {
+ ReaderConfigurationData conf = new ReaderConfigurationData<>();
+ conf.setTopicName(topic);
+ conf.setSubscriptionName(COMPACTION_SUBSCRIPTION);
+ conf.setStartMessageId(MessageId.earliest);
+ conf.setStartMessageFromRollbackDurationInSec(0);
+ conf.setReadCompacted(true);
+ conf.setSubscriptionMode(SubscriptionMode.Durable);
+ conf.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+ conf.setCryptoKeyReader(cryptoKeyReader);
+ return new CompactionReaderImpl(client, conf, client.externalExecutorProvider(), consumerFuture, schema);
+ }
+
+
+ @Override
+ public Message<T> readNext() throws PulsarClientException {
+ return consumer.receive();
+ }
+
+ @Override
+ public Message<T> readNext(int timeout, TimeUnit unit) throws PulsarClientException {
+ return consumer.receive(timeout, unit);
+ }
+
+ @Override
+ public CompletableFuture<Message<T>> readNextAsync() {
+ CompletableFuture<Message<T>> receiveFuture = consumer.receiveAsync();
+ return receiveFuture;
Review Comment:
```suggestion
return consumer.receiveAsync();
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/CompactionReaderImpl.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.client.impl;
+
+import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.api.proto.CommandAck;
+
+/**
+ * An extended ReaderImpl used for StrategicTwoPhaseCompactor.
+ * The compaction consumer subscription is durable and consumes compacted messages from the earliest position.
+ * It does not acknowledge the message after each read. (needs to call acknowledgeCumulativeAsync to ack messages.)
+ */
+@Slf4j
+public class CompactionReaderImpl<T> extends ReaderImpl<T> {
+
+ ConsumerBase<T> consumer;
+
+ ReaderConfigurationData<T> readerConfiguration;
+ private CompactionReaderImpl(PulsarClientImpl client, ReaderConfigurationData<T> readerConfiguration,
+ ExecutorProvider executorProvider, CompletableFuture<Consumer<T>> consumerFuture,
+ Schema<T> schema) {
+ super(client, readerConfiguration, executorProvider, consumerFuture, schema);
+ this.readerConfiguration = readerConfiguration;
+ this.consumer = getConsumer();
+ }
+
+ public static <T> CompactionReaderImpl<T> create(PulsarClientImpl client, Schema<T> schema, String topic,
+ CompletableFuture<Consumer<T>> consumerFuture,
+ CryptoKeyReader cryptoKeyReader) {
+ ReaderConfigurationData conf = new ReaderConfigurationData<>();
+ conf.setTopicName(topic);
+ conf.setSubscriptionName(COMPACTION_SUBSCRIPTION);
+ conf.setStartMessageId(MessageId.earliest);
+ conf.setStartMessageFromRollbackDurationInSec(0);
+ conf.setReadCompacted(true);
+ conf.setSubscriptionMode(SubscriptionMode.Durable);
+ conf.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+ conf.setCryptoKeyReader(cryptoKeyReader);
+ return new CompactionReaderImpl(client, conf, client.externalExecutorProvider(), consumerFuture, schema);
+ }
+
+
+ @Override
+ public Message<T> readNext() throws PulsarClientException {
+ return consumer.receive();
+ }
+
+ @Override
+ public Message<T> readNext(int timeout, TimeUnit unit) throws PulsarClientException {
+ return consumer.receive(timeout, unit);
+ }
+
+ @Override
+ public CompletableFuture<Message<T>> readNextAsync() {
+ CompletableFuture<Message<T>> receiveFuture = consumer.receiveAsync();
+ return receiveFuture;
+ }
+
+ public CompletableFuture<MessageId> getLastMessageIdAsync() {
+ return consumer.getLastMessageIdAsync();
+ }
+
+ public CompletableFuture<Void> acknowledgeCumulativeAsync(MessageId messageId, Map<String, Long> properties) {
+ return consumer.doAcknowledgeWithTxn(messageId, CommandAck.AckType.Cumulative, properties, null);
Review Comment:
Why do we use `doAcknowledgeWithTxn`? Can we use `doAcknowledge` instead?
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/CompactionReaderImplTest.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.client.impl;
+
+import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import com.google.common.collect.Sets;
+import java.util.concurrent.CompletableFuture;
+import lombok.Cleanup;
+import org.apache.commons.lang.reflect.FieldUtils;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-impl")
+public class CompactionReaderImplTest extends MockedPulsarServiceBaseTest {
+
+ @BeforeMethod
+ @Override
+ public void setup() throws Exception {
+ super.internalSetup();
+ admin.clusters().createCluster("test",
+ ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
+ admin.tenants().createTenant("my-property",
+ new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
+ admin.namespaces().createNamespace("my-property/my-ns", Sets.newHashSet("test"));
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ public void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Test
+ public void test() throws Exception {
+
+ String topic = "persistent://my-property/my-ns/my-compact-topic";
+ int numKeys = 5;
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();
+ for (int i = 0; i < numKeys; i++) {
+ producer.newMessage().key("key:" + i).value("value" + i).send();
+ }
+
+ @Cleanup
+ CompactionReaderImpl<String> reader = CompactionReaderImpl
+ .create((PulsarClientImpl) pulsarClient, Schema.STRING, topic, new CompletableFuture(), null);
+
+ ConsumerBase consumerBase = spy(reader.getConsumer());
+ org.apache.commons.lang3.reflect.FieldUtils.writeDeclaredField(
Review Comment:
nit: Remove the full package name.
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -0,0 +1,433 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.impl.BatchMessageIdImpl;
+import org.apache.pulsar.client.impl.CompactionReaderImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.RawBatchMessageContainerImpl;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects valid message(defined in the TopicCompactionStrategy.isValid())
+ * for each key in the topic. Then, the second pass writes these values
+ * to a ledger.
+ *
+ * <p>As the first pass caches the entire message(not just offset) for each key into a map,
+ * this compaction could be memory intensive if the message payload is large.
+ */
+public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
+ private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
+ private static final int MAX_OUTSTANDING = 500;
+ private final Duration phaseOneLoopReadTimeout;
+ private final RawBatchMessageContainerImpl batchMessageContainer;
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler,
+ int maxNumMessagesInBatch) {
+ super(conf, pulsar, bk, scheduler);
+ batchMessageContainer = new RawBatchMessageContainerImpl(maxNumMessagesInBatch);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+ }
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this(conf, pulsar, bk, scheduler, -1);
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy) {
+ return compact(topic, strategy, null);
+ }
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy,
+ CryptoKeyReader cryptoKeyReader) {
+ CompletableFuture<Consumer<T>> consumerFuture = new CompletableFuture<>();
+ if (cryptoKeyReader != null) {
+ batchMessageContainer.setCryptoKeyReader(cryptoKeyReader);
+ }
+ CompactionReaderImpl reader = CompactionReaderImpl.create(
+ (PulsarClientImpl) pulsar, strategy.getSchema(), topic, consumerFuture, cryptoKeyReader);
+
+ return consumerFuture.thenComposeAsync(__ -> compactAndCloseReader(reader, strategy), scheduler);
+ }
+
+ <T> CompletableFuture<Long> doCompaction(Reader<T> reader, TopicCompactionStrategy strategy) {
+
+ if (!(reader instanceof CompactionReaderImpl<T>)) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("reader has to be DelayedAckReaderImpl"));
+ }
+ return reader.hasMessageAvailableAsync()
+ .thenCompose(available -> {
+ if (available) {
+ return phaseOne(reader, strategy)
+ .thenCompose((result) -> phaseTwo(result, reader, bk));
+ } else {
+ log.info("Skip compaction of the empty topic {}", reader.getTopic());
+ return CompletableFuture.completedFuture(-1L);
+ }
+ });
+ }
+
+ <T> CompletableFuture<Long> compactAndCloseReader(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ mxBean.addCompactionStartOp(reader.getTopic());
+ doCompaction(reader, strategy).whenComplete(
+ (ledgerId, exception) -> {
+ log.info("Completed doCompaction ledgerId:{}", ledgerId);
+ reader.closeAsync().whenComplete((v, exception2) -> {
+ if (exception2 != null) {
+ log.warn("Error closing reader handle {}, ignoring", reader, exception2);
+ }
+ if (exception != null) {
+ // complete with original exception
+ mxBean.addCompactionEndOp(reader.getTopic(), false);
+ promise.completeExceptionally(exception);
+ } else {
+ mxBean.addCompactionEndOp(reader.getTopic(), true);
+ promise.complete(ledgerId);
+ }
+ });
+ });
+ return promise;
+ }
+
+ private <T> boolean doCompactMessage(Message<T> msg, PhaseOneResult<T> result, TopicCompactionStrategy strategy) {
+ Map<String, Message<T>> cache = result.cache;
+ String key = msg.getKey();
+
+ if (key == null) {
+ msg.release();
+ return true;
+ }
+ T val = msg.getValue();
+ Message<T> prev = cache.get(key);
+ T prevVal = prev == null ? null : prev.getValue();
+
+ if (strategy.isValid(prevVal, val)) {
+ if (val != null && msg.size() > 0) {
+ cache.remove(key); // to reorder
+ cache.put(key, msg);
+ } else {
+ cache.remove(key);
+ msg.release();
+ }
+
+ if (prev != null) {
+ prev.release();
+ }
+
+ result.validCompactionCount.incrementAndGet();
+ return true;
+ } else {
+ msg.release();
+ result.invalidCompactionCount.incrementAndGet();
+ return false;
+ }
+
+ }
+
+ private static class PhaseOneResult<T> {
+ MessageId firstId;
+ //MessageId to; // last undeleted messageId
Review Comment:
This field is not used?
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/CompactionReaderImpl.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.client.impl;
+
+import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.api.proto.CommandAck;
+
+/**
+ * An extended ReaderImpl used for StrategicTwoPhaseCompactor.
+ * The compaction consumer subscription is durable and consumes compacted messages from the earliest position.
+ * It does not acknowledge the message after each read. (needs to call acknowledgeCumulativeAsync to ack messages.)
+ */
+@Slf4j
+public class CompactionReaderImpl<T> extends ReaderImpl<T> {
+
+ ConsumerBase<T> consumer;
+
+ ReaderConfigurationData<T> readerConfiguration;
+ private CompactionReaderImpl(PulsarClientImpl client, ReaderConfigurationData<T> readerConfiguration,
+ ExecutorProvider executorProvider, CompletableFuture<Consumer<T>> consumerFuture,
+ Schema<T> schema) {
+ super(client, readerConfiguration, executorProvider, consumerFuture, schema);
+ this.readerConfiguration = readerConfiguration;
+ this.consumer = getConsumer();
+ }
+
+ public static <T> CompactionReaderImpl<T> create(PulsarClientImpl client, Schema<T> schema, String topic,
+ CompletableFuture<Consumer<T>> consumerFuture,
+ CryptoKeyReader cryptoKeyReader) {
+ ReaderConfigurationData conf = new ReaderConfigurationData<>();
+ conf.setTopicName(topic);
+ conf.setSubscriptionName(COMPACTION_SUBSCRIPTION);
+ conf.setStartMessageId(MessageId.earliest);
+ conf.setStartMessageFromRollbackDurationInSec(0);
+ conf.setReadCompacted(true);
+ conf.setSubscriptionMode(SubscriptionMode.Durable);
+ conf.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+ conf.setCryptoKeyReader(cryptoKeyReader);
+ return new CompactionReaderImpl(client, conf, client.externalExecutorProvider(), consumerFuture, schema);
Review Comment:
```suggestion
return new CompactionReaderImpl<>(client, conf, client.externalExecutorProvider(), consumerFuture, schema);
```
--
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 commented on a diff in pull request #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1030366755
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -0,0 +1,434 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.impl.BatchMessageIdImpl;
+import org.apache.pulsar.client.impl.CompactionReaderImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.RawBatchMessageContainerImpl;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects valid message(defined in the TopicCompactionStrategy.isValid())
+ * for each key in the topic. Then, the second pass writes these values
+ * to a ledger.
+ *
+ * <p>As the first pass caches the entire message(not just offset) for each key into a map,
+ * this compaction could be memory intensive if the message payload is large.
+ */
+public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
+ private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
+ private static final int MAX_OUTSTANDING = 500;
+ private final Duration phaseOneLoopReadTimeout;
+ private final RawBatchMessageContainerImpl batchMessageContainer;
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler,
+ int maxNumMessagesInBatch) {
+ super(conf, pulsar, bk, scheduler);
+ batchMessageContainer = new RawBatchMessageContainerImpl(maxNumMessagesInBatch);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+ }
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this(conf, pulsar, bk, scheduler, -1);
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy) {
+ return compact(topic, strategy, null);
+ }
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy,
+ CryptoKeyReader cryptoKeyReader) {
+ CompletableFuture<Consumer<T>> consumerFuture = new CompletableFuture<>();
+ if (cryptoKeyReader != null) {
+ batchMessageContainer.setCryptoKeyReader(cryptoKeyReader);
+ }
+ CompactionReaderImpl reader = CompactionReaderImpl.create(
+ (PulsarClientImpl) pulsar, strategy.getSchema(), topic, consumerFuture, cryptoKeyReader);
+
+ return consumerFuture.thenComposeAsync(__ -> compactAndCloseReader(reader, strategy), scheduler);
+ }
+
+ <T> CompletableFuture<Long> doCompaction(Reader<T> reader, TopicCompactionStrategy strategy) {
+
+ if (!(reader instanceof CompactionReaderImpl<T>)) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("reader has to be DelayedAckReaderImpl"));
+ }
+ return reader.hasMessageAvailableAsync()
Review Comment:
Sorry, I missed this part when reviewing the proposal.
--
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 #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1037888292
##########
pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicCompactionStrategy.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.common.topics;
+
+import org.apache.pulsar.client.api.Schema;
+
+/**
+ * Defines a custom strategy to compact messages in a topic.
+ * This strategy can be passed to Topic Compactor and Table View to compact messages in a custom way.
+ *
+ * Examples:
+ *
+ * TopicCompactionStrategy strategy = new MyTopicCompactionStrategy();
+ *
+ * // Run topic compaction by the compaction strategy.
+ * // While compacting messages for each key,
+ * // it will choose messages only if TopicCompactionStrategy.shouldKeepLeft(prev, cur) returns false.
+ * StrategicTwoPhaseCompactor compactor = new StrategicTwoPhaseCompactor(...);
+ * compactor.compact(topic, strategy);
+ *
+ * // Run table view by the compaction strategy.
+ * // While updating messages in the table view <key,value> map,
+ * // it will choose messages only if TopicCompactionStrategy.shouldKeepLeft(prev, cur) returns false.
+ * TableView tableView = pulsar.getClient().newTableViewBuilder(strategy.getSchema())
+ * .topic(topic)
+ * .loadConf(Map.of(
+ * "topicCompactionStrategy", strategy.getClass().getCanonicalName()))
+ * .create();
+ */
+public interface TopicCompactionStrategy<T> {
+
+ /**
+ * Returns the schema object for this strategy.
+ * @return
+ */
+ Schema<T> getSchema();
+ /**
+ * Tests if the compaction needs to keep the left(previous message)
+ * compared to the right(current message) for the same key.
+ *
+ * @param prev previous message value
+ * @param cur current message value
+ * @return True if it needs to keep the previous message and ignore the current message. Otherwise, False.
+ */
+ boolean shouldKeepLeft(T prev, T cur);
Review Comment:
The reason why we have keepLeft and merge is that we dont want to update cache if keepLeft returns true.
With merge api only, it is unclear whether the return does not need to update the cache.
--
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 #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1033154830
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -0,0 +1,434 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.impl.BatchMessageIdImpl;
+import org.apache.pulsar.client.impl.CompactionReaderImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.RawBatchMessageContainerImpl;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects valid message(defined in the TopicCompactionStrategy.isValid())
+ * for each key in the topic. Then, the second pass writes these values
+ * to a ledger.
+ *
+ * <p>As the first pass caches the entire message(not just offset) for each key into a map,
+ * this compaction could be memory intensive if the message payload is large.
+ */
+public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
+ private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
+ private static final int MAX_OUTSTANDING = 500;
+ private final Duration phaseOneLoopReadTimeout;
+ private final RawBatchMessageContainerImpl batchMessageContainer;
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler,
+ int maxNumMessagesInBatch) {
+ super(conf, pulsar, bk, scheduler);
+ batchMessageContainer = new RawBatchMessageContainerImpl(maxNumMessagesInBatch);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+ }
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this(conf, pulsar, bk, scheduler, -1);
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy) {
+ return compact(topic, strategy, null);
+ }
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy,
+ CryptoKeyReader cryptoKeyReader) {
+ CompletableFuture<Consumer<T>> consumerFuture = new CompletableFuture<>();
+ if (cryptoKeyReader != null) {
+ batchMessageContainer.setCryptoKeyReader(cryptoKeyReader);
+ }
+ CompactionReaderImpl reader = CompactionReaderImpl.create(
+ (PulsarClientImpl) pulsar, strategy.getSchema(), topic, consumerFuture, cryptoKeyReader);
+
+ return consumerFuture.thenComposeAsync(__ -> compactAndCloseReader(reader, strategy), scheduler);
+ }
+
+ <T> CompletableFuture<Long> doCompaction(Reader<T> reader, TopicCompactionStrategy strategy) {
+
+ if (!(reader instanceof CompactionReaderImpl<T>)) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("reader has to be DelayedAckReaderImpl"));
+ }
+ return reader.hasMessageAvailableAsync()
Review Comment:
This is indeed a performance concern. This compaction will impact broker performance if the topic to compact is really large. We could further comment warnings that this is more expensive operation than the original compaction.
However, I dont see other ways to merge/compare two messages without deserializing the message payload.
For pip192, we dont expect too many bundles to compact at once( probably less than 100k bundles)
We could discuss more about this in the discussion email 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] heesung-sn commented on a diff in pull request #18195: [improve][broker] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1014537948
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/CompactionReaderImpl.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.client.impl;
+
+import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.api.proto.CommandAck;
+
+/**
+ * An extended ReaderImpl used for StrategicTwoPhaseCompactor.
+ * The compaction consumer subscription is durable and consumes compacted messages from the earliest position.
+ * It does not acknowledge the message after each read. (needs to call acknowledgeCumulativeAsync to ack messages.)
+ */
+@Slf4j
+public class CompactionReaderImpl<T> extends ReaderImpl<T> {
+
+ ConsumerBase<T> consumer;
+
+ ReaderConfigurationData<T> readerConfiguration;
+ private CompactionReaderImpl(PulsarClientImpl client, ReaderConfigurationData<T> readerConfiguration,
+ ExecutorProvider executorProvider, CompletableFuture<Consumer<T>> consumerFuture,
+ Schema<T> schema) {
+ super(client, readerConfiguration, executorProvider, consumerFuture, schema);
+ this.readerConfiguration = readerConfiguration;
+ this.consumer = getConsumer();
+ }
+
+ public static <T> CompactionReaderImpl<T> create(PulsarClientImpl client, Schema<T> schema, String topic,
+ CompletableFuture<Consumer<T>> consumerFuture,
+ CryptoKeyReader cryptoKeyReader) {
+ ReaderConfigurationData conf = new ReaderConfigurationData<>();
Review Comment:
updated.
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/CompactionReaderImpl.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.client.impl;
+
+import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.api.proto.CommandAck;
+
+/**
+ * An extended ReaderImpl used for StrategicTwoPhaseCompactor.
+ * The compaction consumer subscription is durable and consumes compacted messages from the earliest position.
+ * It does not acknowledge the message after each read. (needs to call acknowledgeCumulativeAsync to ack messages.)
+ */
+@Slf4j
+public class CompactionReaderImpl<T> extends ReaderImpl<T> {
+
+ ConsumerBase<T> consumer;
+
+ ReaderConfigurationData<T> readerConfiguration;
+ private CompactionReaderImpl(PulsarClientImpl client, ReaderConfigurationData<T> readerConfiguration,
+ ExecutorProvider executorProvider, CompletableFuture<Consumer<T>> consumerFuture,
+ Schema<T> schema) {
+ super(client, readerConfiguration, executorProvider, consumerFuture, schema);
+ this.readerConfiguration = readerConfiguration;
+ this.consumer = getConsumer();
+ }
+
+ public static <T> CompactionReaderImpl<T> create(PulsarClientImpl client, Schema<T> schema, String topic,
+ CompletableFuture<Consumer<T>> consumerFuture,
+ CryptoKeyReader cryptoKeyReader) {
+ ReaderConfigurationData conf = new ReaderConfigurationData<>();
+ conf.setTopicName(topic);
+ conf.setSubscriptionName(COMPACTION_SUBSCRIPTION);
+ conf.setStartMessageId(MessageId.earliest);
+ conf.setStartMessageFromRollbackDurationInSec(0);
+ conf.setReadCompacted(true);
+ conf.setSubscriptionMode(SubscriptionMode.Durable);
+ conf.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+ conf.setCryptoKeyReader(cryptoKeyReader);
+ return new CompactionReaderImpl(client, conf, client.externalExecutorProvider(), consumerFuture, schema);
Review Comment:
updated.
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/CompactionReaderImpl.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.client.impl;
+
+import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.api.proto.CommandAck;
+
+/**
+ * An extended ReaderImpl used for StrategicTwoPhaseCompactor.
+ * The compaction consumer subscription is durable and consumes compacted messages from the earliest position.
+ * It does not acknowledge the message after each read. (needs to call acknowledgeCumulativeAsync to ack messages.)
+ */
+@Slf4j
+public class CompactionReaderImpl<T> extends ReaderImpl<T> {
+
+ ConsumerBase<T> consumer;
+
+ ReaderConfigurationData<T> readerConfiguration;
+ private CompactionReaderImpl(PulsarClientImpl client, ReaderConfigurationData<T> readerConfiguration,
+ ExecutorProvider executorProvider, CompletableFuture<Consumer<T>> consumerFuture,
+ Schema<T> schema) {
+ super(client, readerConfiguration, executorProvider, consumerFuture, schema);
+ this.readerConfiguration = readerConfiguration;
+ this.consumer = getConsumer();
+ }
+
+ public static <T> CompactionReaderImpl<T> create(PulsarClientImpl client, Schema<T> schema, String topic,
+ CompletableFuture<Consumer<T>> consumerFuture,
+ CryptoKeyReader cryptoKeyReader) {
+ ReaderConfigurationData conf = new ReaderConfigurationData<>();
+ conf.setTopicName(topic);
+ conf.setSubscriptionName(COMPACTION_SUBSCRIPTION);
+ conf.setStartMessageId(MessageId.earliest);
+ conf.setStartMessageFromRollbackDurationInSec(0);
+ conf.setReadCompacted(true);
+ conf.setSubscriptionMode(SubscriptionMode.Durable);
+ conf.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+ conf.setCryptoKeyReader(cryptoKeyReader);
+ return new CompactionReaderImpl(client, conf, client.externalExecutorProvider(), consumerFuture, schema);
+ }
+
+
+ @Override
+ public Message<T> readNext() throws PulsarClientException {
+ return consumer.receive();
+ }
+
+ @Override
+ public Message<T> readNext(int timeout, TimeUnit unit) throws PulsarClientException {
+ return consumer.receive(timeout, unit);
+ }
+
+ @Override
+ public CompletableFuture<Message<T>> readNextAsync() {
+ CompletableFuture<Message<T>> receiveFuture = consumer.receiveAsync();
+ return receiveFuture;
Review Comment:
updated.
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -0,0 +1,433 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.impl.BatchMessageIdImpl;
+import org.apache.pulsar.client.impl.CompactionReaderImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.RawBatchMessageContainerImpl;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects valid message(defined in the TopicCompactionStrategy.isValid())
+ * for each key in the topic. Then, the second pass writes these values
+ * to a ledger.
+ *
+ * <p>As the first pass caches the entire message(not just offset) for each key into a map,
+ * this compaction could be memory intensive if the message payload is large.
+ */
+public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
+ private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
+ private static final int MAX_OUTSTANDING = 500;
+ private final Duration phaseOneLoopReadTimeout;
+ private final RawBatchMessageContainerImpl batchMessageContainer;
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler,
+ int maxNumMessagesInBatch) {
+ super(conf, pulsar, bk, scheduler);
+ batchMessageContainer = new RawBatchMessageContainerImpl(maxNumMessagesInBatch);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+ }
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this(conf, pulsar, bk, scheduler, -1);
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy) {
+ return compact(topic, strategy, null);
+ }
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy,
+ CryptoKeyReader cryptoKeyReader) {
+ CompletableFuture<Consumer<T>> consumerFuture = new CompletableFuture<>();
+ if (cryptoKeyReader != null) {
+ batchMessageContainer.setCryptoKeyReader(cryptoKeyReader);
+ }
+ CompactionReaderImpl reader = CompactionReaderImpl.create(
+ (PulsarClientImpl) pulsar, strategy.getSchema(), topic, consumerFuture, cryptoKeyReader);
+
+ return consumerFuture.thenComposeAsync(__ -> compactAndCloseReader(reader, strategy), scheduler);
+ }
+
+ <T> CompletableFuture<Long> doCompaction(Reader<T> reader, TopicCompactionStrategy strategy) {
+
+ if (!(reader instanceof CompactionReaderImpl<T>)) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("reader has to be DelayedAckReaderImpl"));
+ }
+ return reader.hasMessageAvailableAsync()
+ .thenCompose(available -> {
+ if (available) {
+ return phaseOne(reader, strategy)
+ .thenCompose((result) -> phaseTwo(result, reader, bk));
+ } else {
+ log.info("Skip compaction of the empty topic {}", reader.getTopic());
+ return CompletableFuture.completedFuture(-1L);
+ }
+ });
+ }
+
+ <T> CompletableFuture<Long> compactAndCloseReader(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ mxBean.addCompactionStartOp(reader.getTopic());
+ doCompaction(reader, strategy).whenComplete(
+ (ledgerId, exception) -> {
+ log.info("Completed doCompaction ledgerId:{}", ledgerId);
+ reader.closeAsync().whenComplete((v, exception2) -> {
+ if (exception2 != null) {
+ log.warn("Error closing reader handle {}, ignoring", reader, exception2);
+ }
+ if (exception != null) {
+ // complete with original exception
+ mxBean.addCompactionEndOp(reader.getTopic(), false);
+ promise.completeExceptionally(exception);
+ } else {
+ mxBean.addCompactionEndOp(reader.getTopic(), true);
+ promise.complete(ledgerId);
+ }
+ });
+ });
+ return promise;
+ }
+
+ private <T> boolean doCompactMessage(Message<T> msg, PhaseOneResult<T> result, TopicCompactionStrategy strategy) {
+ Map<String, Message<T>> cache = result.cache;
+ String key = msg.getKey();
+
+ if (key == null) {
+ msg.release();
+ return true;
+ }
+ T val = msg.getValue();
+ Message<T> prev = cache.get(key);
+ T prevVal = prev == null ? null : prev.getValue();
+
+ if (strategy.isValid(prevVal, val)) {
+ if (val != null && msg.size() > 0) {
+ cache.remove(key); // to reorder
+ cache.put(key, msg);
+ } else {
+ cache.remove(key);
+ msg.release();
+ }
+
+ if (prev != null) {
+ prev.release();
+ }
+
+ result.validCompactionCount.incrementAndGet();
+ return true;
+ } else {
+ msg.release();
+ result.invalidCompactionCount.incrementAndGet();
+ return false;
+ }
+
+ }
+
+ private static class PhaseOneResult<T> {
+ MessageId firstId;
+ //MessageId to; // last undeleted messageId
Review Comment:
not used. removed.
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -0,0 +1,433 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.impl.BatchMessageIdImpl;
+import org.apache.pulsar.client.impl.CompactionReaderImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.RawBatchMessageContainerImpl;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects valid message(defined in the TopicCompactionStrategy.isValid())
+ * for each key in the topic. Then, the second pass writes these values
+ * to a ledger.
+ *
+ * <p>As the first pass caches the entire message(not just offset) for each key into a map,
+ * this compaction could be memory intensive if the message payload is large.
+ */
+public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
+ private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
+ private static final int MAX_OUTSTANDING = 500;
+ private final Duration phaseOneLoopReadTimeout;
+ private final RawBatchMessageContainerImpl batchMessageContainer;
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler,
+ int maxNumMessagesInBatch) {
+ super(conf, pulsar, bk, scheduler);
+ batchMessageContainer = new RawBatchMessageContainerImpl(maxNumMessagesInBatch);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+ }
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this(conf, pulsar, bk, scheduler, -1);
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy) {
+ return compact(topic, strategy, null);
+ }
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy,
+ CryptoKeyReader cryptoKeyReader) {
+ CompletableFuture<Consumer<T>> consumerFuture = new CompletableFuture<>();
+ if (cryptoKeyReader != null) {
+ batchMessageContainer.setCryptoKeyReader(cryptoKeyReader);
+ }
+ CompactionReaderImpl reader = CompactionReaderImpl.create(
+ (PulsarClientImpl) pulsar, strategy.getSchema(), topic, consumerFuture, cryptoKeyReader);
+
+ return consumerFuture.thenComposeAsync(__ -> compactAndCloseReader(reader, strategy), scheduler);
+ }
+
+ <T> CompletableFuture<Long> doCompaction(Reader<T> reader, TopicCompactionStrategy strategy) {
+
+ if (!(reader instanceof CompactionReaderImpl<T>)) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("reader has to be DelayedAckReaderImpl"));
+ }
+ return reader.hasMessageAvailableAsync()
+ .thenCompose(available -> {
+ if (available) {
+ return phaseOne(reader, strategy)
+ .thenCompose((result) -> phaseTwo(result, reader, bk));
+ } else {
+ log.info("Skip compaction of the empty topic {}", reader.getTopic());
+ return CompletableFuture.completedFuture(-1L);
+ }
+ });
+ }
+
+ <T> CompletableFuture<Long> compactAndCloseReader(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ mxBean.addCompactionStartOp(reader.getTopic());
+ doCompaction(reader, strategy).whenComplete(
+ (ledgerId, exception) -> {
+ log.info("Completed doCompaction ledgerId:{}", ledgerId);
+ reader.closeAsync().whenComplete((v, exception2) -> {
+ if (exception2 != null) {
+ log.warn("Error closing reader handle {}, ignoring", reader, exception2);
+ }
+ if (exception != null) {
+ // complete with original exception
+ mxBean.addCompactionEndOp(reader.getTopic(), false);
+ promise.completeExceptionally(exception);
+ } else {
+ mxBean.addCompactionEndOp(reader.getTopic(), true);
+ promise.complete(ledgerId);
+ }
+ });
+ });
+ return promise;
+ }
+
+ private <T> boolean doCompactMessage(Message<T> msg, PhaseOneResult<T> result, TopicCompactionStrategy strategy) {
Review Comment:
updated.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java:
##########
@@ -164,8 +164,8 @@ public Message<T> readNext() throws PulsarClientException {
@Override
public Message<T> readNext(int timeout, TimeUnit unit) throws PulsarClientException {
- Message<T> msg = consumer.receive(timeout, unit);
+ Message<T> msg = consumer.receive(timeout, unit);
Review Comment:
removed.
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/CompactionReaderImplTest.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.client.impl;
+
+import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import com.google.common.collect.Sets;
+import java.util.concurrent.CompletableFuture;
+import lombok.Cleanup;
+import org.apache.commons.lang.reflect.FieldUtils;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-impl")
+public class CompactionReaderImplTest extends MockedPulsarServiceBaseTest {
+
+ @BeforeMethod
+ @Override
+ public void setup() throws Exception {
+ super.internalSetup();
+ admin.clusters().createCluster("test",
+ ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
+ admin.tenants().createTenant("my-property",
+ new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
+ admin.namespaces().createNamespace("my-property/my-ns", Sets.newHashSet("test"));
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ public void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Test
+ public void test() throws Exception {
+
+ String topic = "persistent://my-property/my-ns/my-compact-topic";
+ int numKeys = 5;
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();
+ for (int i = 0; i < numKeys; i++) {
+ producer.newMessage().key("key:" + i).value("value" + i).send();
+ }
+
+ @Cleanup
+ CompactionReaderImpl<String> reader = CompactionReaderImpl
+ .create((PulsarClientImpl) pulsarClient, Schema.STRING, topic, new CompletableFuture(), null);
+
+ ConsumerBase consumerBase = spy(reader.getConsumer());
+ org.apache.commons.lang3.reflect.FieldUtils.writeDeclaredField(
Review Comment:
removed.
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -0,0 +1,433 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.impl.BatchMessageIdImpl;
+import org.apache.pulsar.client.impl.CompactionReaderImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.RawBatchMessageContainerImpl;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects valid message(defined in the TopicCompactionStrategy.isValid())
+ * for each key in the topic. Then, the second pass writes these values
+ * to a ledger.
+ *
+ * <p>As the first pass caches the entire message(not just offset) for each key into a map,
+ * this compaction could be memory intensive if the message payload is large.
+ */
+public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
+ private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
+ private static final int MAX_OUTSTANDING = 500;
+ private final Duration phaseOneLoopReadTimeout;
+ private final RawBatchMessageContainerImpl batchMessageContainer;
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler,
+ int maxNumMessagesInBatch) {
+ super(conf, pulsar, bk, scheduler);
+ batchMessageContainer = new RawBatchMessageContainerImpl(maxNumMessagesInBatch);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+ }
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this(conf, pulsar, bk, scheduler, -1);
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy) {
+ return compact(topic, strategy, null);
+ }
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy,
+ CryptoKeyReader cryptoKeyReader) {
+ CompletableFuture<Consumer<T>> consumerFuture = new CompletableFuture<>();
+ if (cryptoKeyReader != null) {
+ batchMessageContainer.setCryptoKeyReader(cryptoKeyReader);
+ }
+ CompactionReaderImpl reader = CompactionReaderImpl.create(
+ (PulsarClientImpl) pulsar, strategy.getSchema(), topic, consumerFuture, cryptoKeyReader);
+
+ return consumerFuture.thenComposeAsync(__ -> compactAndCloseReader(reader, strategy), scheduler);
+ }
+
+ <T> CompletableFuture<Long> doCompaction(Reader<T> reader, TopicCompactionStrategy strategy) {
+
+ if (!(reader instanceof CompactionReaderImpl<T>)) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("reader has to be DelayedAckReaderImpl"));
+ }
+ return reader.hasMessageAvailableAsync()
+ .thenCompose(available -> {
+ if (available) {
+ return phaseOne(reader, strategy)
+ .thenCompose((result) -> phaseTwo(result, reader, bk));
+ } else {
+ log.info("Skip compaction of the empty topic {}", reader.getTopic());
+ return CompletableFuture.completedFuture(-1L);
+ }
+ });
+ }
+
+ <T> CompletableFuture<Long> compactAndCloseReader(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ mxBean.addCompactionStartOp(reader.getTopic());
+ doCompaction(reader, strategy).whenComplete(
+ (ledgerId, exception) -> {
+ log.info("Completed doCompaction ledgerId:{}", ledgerId);
+ reader.closeAsync().whenComplete((v, exception2) -> {
+ if (exception2 != null) {
+ log.warn("Error closing reader handle {}, ignoring", reader, exception2);
+ }
+ if (exception != null) {
+ // complete with original exception
+ mxBean.addCompactionEndOp(reader.getTopic(), false);
+ promise.completeExceptionally(exception);
+ } else {
+ mxBean.addCompactionEndOp(reader.getTopic(), true);
+ promise.complete(ledgerId);
+ }
+ });
+ });
+ return promise;
+ }
+
+ private <T> boolean doCompactMessage(Message<T> msg, PhaseOneResult<T> result, TopicCompactionStrategy strategy) {
+ Map<String, Message<T>> cache = result.cache;
+ String key = msg.getKey();
+
+ if (key == null) {
+ msg.release();
+ return true;
+ }
+ T val = msg.getValue();
+ Message<T> prev = cache.get(key);
+ T prevVal = prev == null ? null : prev.getValue();
+
+ if (strategy.isValid(prevVal, val)) {
+ if (val != null && msg.size() > 0) {
+ cache.remove(key); // to reorder
+ cache.put(key, msg);
+ } else {
+ cache.remove(key);
+ msg.release();
+ }
+
+ if (prev != null) {
+ prev.release();
+ }
+
+ result.validCompactionCount.incrementAndGet();
+ return true;
+ } else {
+ msg.release();
+ result.invalidCompactionCount.incrementAndGet();
+ return false;
+ }
+
+ }
+
+ private static class PhaseOneResult<T> {
+ MessageId firstId;
+ //MessageId to; // last undeleted messageId
+ MessageId lastId; // last read messageId
+ Map<String, Message<T>> cache;
+
+ AtomicInteger invalidCompactionCount;
+
+ AtomicInteger validCompactionCount;
+
+ AtomicInteger numReadMessages;
+
+ String topic;
+
+ PhaseOneResult(String topic) {
+ this.topic = topic;
+ cache = new LinkedHashMap<>();
+ invalidCompactionCount = new AtomicInteger();
+ validCompactionCount = new AtomicInteger();
+ numReadMessages = new AtomicInteger();
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "{Topic:%s, firstId:%s, lastId:%s, cache.size:%d, "
+ + "invalidCompactionCount:%d, validCompactionCount:%d, numReadMessages:%d}",
+ topic,
+ firstId != null ? firstId.toString() : "",
+ lastId != null ? lastId.toString() : "",
+ cache.size(),
+ invalidCompactionCount.get(),
+ validCompactionCount.get(),
+ numReadMessages.get());
+ }
+ }
+
+
+ private <T> CompletableFuture<PhaseOneResult> phaseOne(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<PhaseOneResult> promise = new CompletableFuture<>();
+ PhaseOneResult<T> result = new PhaseOneResult(reader.getTopic());
+
+ ((CompactionReaderImpl<T>) reader).getLastMessageIdAsync()
+ .thenAccept(lastMessageId -> {
+ log.info("Commencing phase one of compaction for {}, reading to {}",
+ reader.getTopic(), lastMessageId);
+ result.lastId = copyMessageId(lastMessageId);
+ phaseOneLoop(reader, promise, result, strategy);
+ }).exceptionally(ex -> {
+ promise.completeExceptionally(ex);
+ return null;
+ });
+
+ return promise;
+
+ }
+
+ private static MessageId copyMessageId(MessageId msgId) {
+ if (msgId instanceof BatchMessageIdImpl) {
+ BatchMessageIdImpl tempId = (BatchMessageIdImpl) msgId;
+ return new BatchMessageIdImpl(tempId);
+ } else if (msgId instanceof MessageIdImpl) {
+ MessageIdImpl tempId = (MessageIdImpl) msgId;
+ return new MessageIdImpl(tempId.getLedgerId(), tempId.getEntryId(),
+ tempId.getPartitionIndex());
+ } else {
+ throw new IllegalStateException("Unknown lastMessageId type");
+ }
+ }
+
+ private <T> void phaseOneLoop(Reader<T> reader, CompletableFuture<PhaseOneResult> promise,
+ PhaseOneResult<T> result, TopicCompactionStrategy strategy) {
Review Comment:
updated
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java:
##########
@@ -51,13 +52,15 @@
private final List<BiConsumer<String, T>> listeners;
private final ReentrantLock listenersMutex;
+ private TopicCompactionStrategy<T> compactionStrategy;
TableViewImpl(PulsarClientImpl client, Schema<T> schema, TableViewConfigurationData conf) {
this.conf = conf;
this.data = new ConcurrentHashMap<>();
this.immutableData = Collections.unmodifiableMap(data);
this.listeners = new ArrayList<>();
this.listenersMutex = new ReentrantLock();
+ this.compactionStrategy = TopicCompactionStrategy.load(conf.getTopicCompactionStrategy());
Review Comment:
Yes. It is covered in TopicCompactionStrategyTest and StrategicCompactionTest.testNumericOrderCompaction().
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/CompactionReaderImpl.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.client.impl;
+
+import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.api.proto.CommandAck;
+
+/**
+ * An extended ReaderImpl used for StrategicTwoPhaseCompactor.
+ * The compaction consumer subscription is durable and consumes compacted messages from the earliest position.
+ * It does not acknowledge the message after each read. (needs to call acknowledgeCumulativeAsync to ack messages.)
+ */
+@Slf4j
+public class CompactionReaderImpl<T> extends ReaderImpl<T> {
+
+ ConsumerBase<T> consumer;
+
+ ReaderConfigurationData<T> readerConfiguration;
+ private CompactionReaderImpl(PulsarClientImpl client, ReaderConfigurationData<T> readerConfiguration,
+ ExecutorProvider executorProvider, CompletableFuture<Consumer<T>> consumerFuture,
+ Schema<T> schema) {
+ super(client, readerConfiguration, executorProvider, consumerFuture, schema);
+ this.readerConfiguration = readerConfiguration;
+ this.consumer = getConsumer();
+ }
+
+ public static <T> CompactionReaderImpl<T> create(PulsarClientImpl client, Schema<T> schema, String topic,
+ CompletableFuture<Consumer<T>> consumerFuture,
+ CryptoKeyReader cryptoKeyReader) {
+ ReaderConfigurationData conf = new ReaderConfigurationData<>();
+ conf.setTopicName(topic);
+ conf.setSubscriptionName(COMPACTION_SUBSCRIPTION);
+ conf.setStartMessageId(MessageId.earliest);
+ conf.setStartMessageFromRollbackDurationInSec(0);
+ conf.setReadCompacted(true);
+ conf.setSubscriptionMode(SubscriptionMode.Durable);
+ conf.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+ conf.setCryptoKeyReader(cryptoKeyReader);
+ return new CompactionReaderImpl(client, conf, client.externalExecutorProvider(), consumerFuture, schema);
+ }
+
+
+ @Override
+ public Message<T> readNext() throws PulsarClientException {
+ return consumer.receive();
+ }
+
+ @Override
+ public Message<T> readNext(int timeout, TimeUnit unit) throws PulsarClientException {
+ return consumer.receive(timeout, unit);
+ }
+
+ @Override
+ public CompletableFuture<Message<T>> readNextAsync() {
+ CompletableFuture<Message<T>> receiveFuture = consumer.receiveAsync();
+ return receiveFuture;
+ }
+
+ public CompletableFuture<MessageId> getLastMessageIdAsync() {
+ return consumer.getLastMessageIdAsync();
+ }
+
+ public CompletableFuture<Void> acknowledgeCumulativeAsync(MessageId messageId, Map<String, Long> properties) {
+ return consumer.doAcknowledgeWithTxn(messageId, CommandAck.AckType.Cumulative, properties, null);
Review Comment:
No good reason. Replaced it with `doAcknowledge`.
--
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 commented on a diff in pull request #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1030365137
##########
pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicCompactionStrategy.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.common.topics;
+
+import org.apache.pulsar.client.api.Schema;
+
+/**
+ * Defines a custom strategy to compact messages in a topic.
+ * This strategy can be passed to Topic Compactor and Table View to compact messages in a custom way.
+ *
+ * Examples:
+ *
+ * TopicCompactionStrategy strategy = new MyTopicCompactionStrategy();
+ *
+ * // Run topic compaction by the compaction strategy.
+ * // While compacting messages for each key,
+ * // it will choose messages only if TopicCompactionStrategy.shouldKeepLeft(prev, cur) returns false.
+ * StrategicTwoPhaseCompactor compactor = new StrategicTwoPhaseCompactor(...);
+ * compactor.compact(topic, strategy);
+ *
+ * // Run table view by the compaction strategy.
+ * // While updating messages in the table view <key,value> map,
+ * // it will choose messages only if TopicCompactionStrategy.shouldKeepLeft(prev, cur) returns false.
+ * TableView tableView = pulsar.getClient().newTableViewBuilder(strategy.getSchema())
+ * .topic(topic)
+ * .loadConf(Map.of(
+ * "topicCompactionStrategy", strategy.getClass().getCanonicalName()))
+ * .create();
+ */
+public interface TopicCompactionStrategy<T> {
+
+ /**
+ * Returns the schema object for this strategy.
+ * @return
+ */
+ Schema<T> getSchema();
+ /**
+ * Tests if the compaction needs to keep the left(previous message)
+ * compared to the right(current message) for the same key.
+ *
+ * @param prev previous message value
+ * @param cur current message value
+ * @return True if it needs to keep the previous message and ignore the current message. Otherwise, False.
+ */
+ boolean shouldKeepLeft(T prev, T cur);
Review Comment:
Sorry, maybe I misunderstand this one. From the proposal, we have `merge()` method.
--
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 #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1028289723
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.client.impl;
+
+import io.netty.buffer.ByteBuf;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.MessageCrypto;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
+import org.apache.pulsar.common.api.EncryptionContext;
+import org.apache.pulsar.common.api.proto.CompressionType;
+import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.compression.CompressionCodecNone;
+import org.apache.pulsar.common.compression.CompressionCodecProvider;
+import org.apache.pulsar.common.protocol.Commands;
+
+/**
+ * A raw batch message container without producer. (Used for StrategicTwoPhaseCompactor)
+ *
+ * incoming single messages:
+ * (k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)
+ *
+ * batched into single batch message:
+ * [(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]
+ */
+public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
Review Comment:
We want to reuse the batching and payload serialization logic from the parent class.
I do not think this PR adds significant complexity to the parent class.
It just checks if the producer is null or not.
We could pass a mock producer if adding the producer null-check is not desirable.
--
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 #18195: [improve][broker] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
Demogorgon314 commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1016633977
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.client.impl;
+
+import io.netty.buffer.ByteBuf;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.MessageCrypto;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
+import org.apache.pulsar.common.api.EncryptionContext;
+import org.apache.pulsar.common.api.proto.CompressionType;
+import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.compression.CompressionCodecNone;
+import org.apache.pulsar.common.compression.CompressionCodecProvider;
+import org.apache.pulsar.common.protocol.Commands;
+
+/**
+ * A raw batch message container without producer. (Used for StrategicTwoPhaseCompactor)
+ *
+ * incoming single messages:
+ * (k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)
+ *
+ * batched into single batch message:
+ * [(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;
+ public RawBatchMessageContainerImpl(int maxNumMessagesInBatch) {
+ super();
+ compressionType = CompressionType.NONE;
+ compressor = new CompressionCodecNone();
+ if (maxNumMessagesInBatch > 0) {
+ this.maxNumMessagesInBatch = maxNumMessagesInBatch;
+ }
+ }
+ private ByteBuf encrypt(ByteBuf compressedPayload) {
+ if (msgCrypto == null) {
+ return compressedPayload;
+ }
+ int maxSize = msgCrypto.getMaxOutputSize(compressedPayload.readableBytes());
+ ByteBuf encryptedPayload = allocator.buffer(maxSize);
+ ByteBuffer targetBuffer = encryptedPayload.nioBuffer(0, maxSize);
+
+ try {
+ msgCrypto.encrypt(encryptionKeys, cryptoKeyReader, () -> messageMetadata,
+ compressedPayload.nioBuffer(), targetBuffer);
+ } catch (PulsarClientException e) {
+ encryptedPayload.release();
+ compressedPayload.release();
+ discard(e);
+ throw new RuntimeException("Failed to encrypt payload", e);
+ }
+ encryptedPayload.writerIndex(targetBuffer.remaining());
+ compressedPayload.release();
+ return encryptedPayload;
+ }
+
+ @Override
+ public ProducerImpl.OpSendMsg createOpSendMsg() {
+ throw new UnsupportedOperationException();
+ }
+ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
+ this.cryptoKeyReader = cryptoKeyReader;
+ }
+
+ public ByteBuf toByteBuf() {
Review Comment:
Can we add some comments to describe which exceptions will throw in this method? Since this method is public, some others develop might use it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui merged pull request #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
codelipenghui merged PR #18195:
URL: https://github.com/apache/pulsar/pull/18195
--
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 #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1033156246
##########
pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicCompactionStrategy.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.common.topics;
+
+import org.apache.pulsar.client.api.Schema;
+
+/**
+ * Defines a custom strategy to compact messages in a topic.
+ * This strategy can be passed to Topic Compactor and Table View to compact messages in a custom way.
+ *
+ * Examples:
+ *
+ * TopicCompactionStrategy strategy = new MyTopicCompactionStrategy();
+ *
+ * // Run topic compaction by the compaction strategy.
+ * // While compacting messages for each key,
+ * // it will choose messages only if TopicCompactionStrategy.shouldKeepLeft(prev, cur) returns false.
+ * StrategicTwoPhaseCompactor compactor = new StrategicTwoPhaseCompactor(...);
+ * compactor.compact(topic, strategy);
+ *
+ * // Run table view by the compaction strategy.
+ * // While updating messages in the table view <key,value> map,
+ * // it will choose messages only if TopicCompactionStrategy.shouldKeepLeft(prev, cur) returns false.
+ * TableView tableView = pulsar.getClient().newTableViewBuilder(strategy.getSchema())
+ * .topic(topic)
+ * .loadConf(Map.of(
+ * "topicCompactionStrategy", strategy.getClass().getCanonicalName()))
+ * .create();
+ */
+public interface TopicCompactionStrategy<T> {
+
+ /**
+ * Returns the schema object for this strategy.
+ * @return
+ */
+ Schema<T> getSchema();
+ /**
+ * Tests if the compaction needs to keep the left(previous message)
+ * compared to the right(current message) for the same key.
+ *
+ * @param prev previous message value
+ * @param cur current message value
+ * @return True if it needs to keep the previous message and ignore the current message. Otherwise, False.
+ */
+ boolean shouldKeepLeft(T prev, T cur);
Review Comment:
merge requires more to construct a new message with the merged values. Since pip192 does not require merge, I dont implement it here in this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: 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 #18195: [improve][broker] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1015805463
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -0,0 +1,433 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.impl.BatchMessageIdImpl;
+import org.apache.pulsar.client.impl.CompactionReaderImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.RawBatchMessageContainerImpl;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects valid message(defined in the TopicCompactionStrategy.isValid())
+ * for each key in the topic. Then, the second pass writes these values
+ * to a ledger.
+ *
+ * <p>As the first pass caches the entire message(not just offset) for each key into a map,
+ * this compaction could be memory intensive if the message payload is large.
+ */
+public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
+ private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
+ private static final int MAX_OUTSTANDING = 500;
+ private final Duration phaseOneLoopReadTimeout;
+ private final RawBatchMessageContainerImpl batchMessageContainer;
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler,
+ int maxNumMessagesInBatch) {
+ super(conf, pulsar, bk, scheduler);
+ batchMessageContainer = new RawBatchMessageContainerImpl(maxNumMessagesInBatch);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+ }
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this(conf, pulsar, bk, scheduler, -1);
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy) {
+ return compact(topic, strategy, null);
+ }
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy,
+ CryptoKeyReader cryptoKeyReader) {
+ CompletableFuture<Consumer<T>> consumerFuture = new CompletableFuture<>();
+ if (cryptoKeyReader != null) {
+ batchMessageContainer.setCryptoKeyReader(cryptoKeyReader);
+ }
+ CompactionReaderImpl reader = CompactionReaderImpl.create(
+ (PulsarClientImpl) pulsar, strategy.getSchema(), topic, consumerFuture, cryptoKeyReader);
+
+ return consumerFuture.thenComposeAsync(__ -> compactAndCloseReader(reader, strategy), scheduler);
+ }
+
+ <T> CompletableFuture<Long> doCompaction(Reader<T> reader, TopicCompactionStrategy strategy) {
+
+ if (!(reader instanceof CompactionReaderImpl<T>)) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("reader has to be DelayedAckReaderImpl"));
+ }
+ return reader.hasMessageAvailableAsync()
+ .thenCompose(available -> {
+ if (available) {
+ return phaseOne(reader, strategy)
+ .thenCompose((result) -> phaseTwo(result, reader, bk));
+ } else {
+ log.info("Skip compaction of the empty topic {}", reader.getTopic());
+ return CompletableFuture.completedFuture(-1L);
+ }
+ });
+ }
+
+ <T> CompletableFuture<Long> compactAndCloseReader(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ mxBean.addCompactionStartOp(reader.getTopic());
+ doCompaction(reader, strategy).whenComplete(
+ (ledgerId, exception) -> {
+ log.info("Completed doCompaction ledgerId:{}", ledgerId);
+ reader.closeAsync().whenComplete((v, exception2) -> {
+ if (exception2 != null) {
+ log.warn("Error closing reader handle {}, ignoring", reader, exception2);
+ }
+ if (exception != null) {
+ // complete with original exception
+ mxBean.addCompactionEndOp(reader.getTopic(), false);
+ promise.completeExceptionally(exception);
+ } else {
+ mxBean.addCompactionEndOp(reader.getTopic(), true);
+ promise.complete(ledgerId);
+ }
+ });
+ });
+ return promise;
+ }
+
+ private <T> boolean doCompactMessage(
+ Message<T> msg, PhaseOneResult<T> result, TopicCompactionStrategy<T> strategy) {
+ Map<String, Message<T>> cache = result.cache;
+ String key = msg.getKey();
+
+ if (key == null) {
+ msg.release();
+ return true;
+ }
+ T val = msg.getValue();
+ Message<T> prev = cache.get(key);
+ T prevVal = prev == null ? null : prev.getValue();
+
+ if (!strategy.shouldKeepLeft(prevVal, val)) {
+ if (val != null && msg.size() > 0) {
+ cache.remove(key); // to reorder
+ cache.put(key, msg);
+ } else {
+ cache.remove(key);
+ msg.release();
+ }
+
+ if (prev != null) {
+ prev.release();
+ }
+
+ result.validCompactionCount.incrementAndGet();
+ return true;
+ } else {
+ msg.release();
+ result.invalidCompactionCount.incrementAndGet();
+ return false;
+ }
+
+ }
+
+ private static class PhaseOneResult<T> {
+ MessageId firstId;
+ MessageId lastId; // last read messageId
+ Map<String, Message<T>> cache;
+
+ AtomicInteger invalidCompactionCount;
+
+ AtomicInteger validCompactionCount;
+
+ AtomicInteger numReadMessages;
+
+ String topic;
+
+ PhaseOneResult(String topic) {
+ this.topic = topic;
+ cache = new LinkedHashMap<>();
+ invalidCompactionCount = new AtomicInteger();
+ validCompactionCount = new AtomicInteger();
+ numReadMessages = new AtomicInteger();
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "{Topic:%s, firstId:%s, lastId:%s, cache.size:%d, "
+ + "invalidCompactionCount:%d, validCompactionCount:%d, numReadMessages:%d}",
+ topic,
+ firstId != null ? firstId.toString() : "",
+ lastId != null ? lastId.toString() : "",
+ cache.size(),
+ invalidCompactionCount.get(),
+ validCompactionCount.get(),
+ numReadMessages.get());
+ }
+ }
+
+
+ private <T> CompletableFuture<PhaseOneResult> phaseOne(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<PhaseOneResult> promise = new CompletableFuture<>();
+ PhaseOneResult<T> result = new PhaseOneResult(reader.getTopic());
+
+ ((CompactionReaderImpl<T>) reader).getLastMessageIdAsync()
+ .thenAccept(lastMessageId -> {
+ log.info("Commencing phase one of compaction for {}, reading to {}",
+ reader.getTopic(), lastMessageId);
+ result.lastId = copyMessageId(lastMessageId);
+ phaseOneLoop(reader, promise, result, strategy);
+ }).exceptionally(ex -> {
+ promise.completeExceptionally(ex);
+ return null;
+ });
+
+ return promise;
+
+ }
+
+ private static MessageId copyMessageId(MessageId msgId) {
+ if (msgId instanceof BatchMessageIdImpl) {
+ BatchMessageIdImpl tempId = (BatchMessageIdImpl) msgId;
+ return new BatchMessageIdImpl(tempId);
+ } else if (msgId instanceof MessageIdImpl) {
+ MessageIdImpl tempId = (MessageIdImpl) msgId;
+ return new MessageIdImpl(tempId.getLedgerId(), tempId.getEntryId(),
+ tempId.getPartitionIndex());
+ } else {
+ throw new IllegalStateException("Unknown lastMessageId type");
+ }
+ }
+
+ private <T> void phaseOneLoop(Reader<T> reader, CompletableFuture<PhaseOneResult> promise,
+ PhaseOneResult<T> result, TopicCompactionStrategy<T> strategy) {
+
+ if (promise.isDone()) {
+ return;
+ }
+
+ CompletableFuture<Message<T>> future = reader.readNextAsync();
+ FutureUtil.addTimeoutHandling(future,
+ phaseOneLoopReadTimeout, scheduler,
+ () -> FutureUtil.createTimeoutException("Timeout", getClass(),
+ "phaseOneLoop(...)"));
+
+ future.thenAcceptAsync(msg -> {
+
+ MessageId id = msg.getMessageId();
+ boolean completed = false;
+ if (result.lastId.compareTo(id) == 0) {
+ completed = true;
+ }
+
+ result.numReadMessages.incrementAndGet();
+ mxBean.addCompactionReadOp(reader.getTopic(), msg.size());
+ if (doCompactMessage(msg, result, strategy)) {
+ mxBean.addCompactionRemovedEvent(reader.getTopic());
+ }
+ //set ids in the result
+ if (result.firstId == null) {
+ result.firstId = copyMessageId(id);
+ log.info("Resetting cursor to firstId:{}", result.firstId);
+ try {
+ reader.seek(result.firstId);
+ } catch (PulsarClientException e) {
+ throw new RuntimeException("Failed to reset the cursor to firstId:" + result.firstId, e);
+ }
+ }
+ if (completed) {
+ promise.complete(result);
+ } else {
+ phaseOneLoop(reader, promise, result, strategy);
+ }
+
+ }, scheduler).exceptionally(ex -> {
+ promise.completeExceptionally(ex);
+ return null;
+ });
+
+ }
+
+ private <T> CompletableFuture<Long> phaseTwo(PhaseOneResult<T> phaseOneResult, Reader<T> reader, BookKeeper bk) {
+ log.info("Completed phase one. Result:{}. ", phaseOneResult);
+ Map<String, byte[]> metadata =
+ LedgerMetadataUtils.buildMetadataForCompactedLedger(
+ phaseOneResult.topic, phaseOneResult.lastId.toByteArray());
+ return createLedger(bk, metadata)
+ .thenCompose((ledger) -> {
+ log.info(
+ "Commencing phase two of compaction for {}, from {} to {}, compacting {} keys to ledger {}",
+ phaseOneResult.topic, phaseOneResult.firstId, phaseOneResult.lastId,
+ phaseOneResult.cache.size(), ledger.getId());
+ return runPhaseTwo(phaseOneResult, reader, ledger, bk);
+ });
+ }
+
+ private <T> CompletableFuture<Long> runPhaseTwo(
+ PhaseOneResult<T> phaseOneResult, Reader<T> reader, LedgerHandle ledger, BookKeeper bk) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ Semaphore outstanding = new Semaphore(MAX_OUTSTANDING);
+ CompletableFuture<Void> loopPromise = new CompletableFuture<>();
+ phaseTwoLoop(phaseOneResult.topic, phaseOneResult.cache.values().iterator(), ledger,
+ outstanding, loopPromise);
+ loopPromise.thenCompose((v) -> {
+ log.info("Flushing batch container numMessagesInBatch:{}",
+ batchMessageContainer.getNumMessagesInBatch());
+ return addToCompactedLedger(ledger, null, reader.getTopic(), outstanding)
+ .whenComplete((res, exception2) -> {
+ if (exception2 != null) {
+ promise.completeExceptionally(exception2);
+ return;
+ }
+ });
+ })
+ .thenCompose(v -> {
+ log.info("Acking ledger id {}", phaseOneResult.firstId);
+ return ((CompactionReaderImpl<T>) reader)
+ .acknowledgeCumulativeAsync(
+ phaseOneResult.lastId, Map.of(COMPACTED_TOPIC_LEDGER_PROPERTY,
+ ledger.getId()));
+ })
+ .thenCompose((v) -> closeLedger(ledger))
+ .whenComplete((v, exception) -> {
+ if (exception != null) {
+ deleteLedger(bk, ledger).whenComplete((res2, exception2) -> {
+ if (exception2 != null) {
+ log.error("Cleanup of ledger {} for failed", ledger, exception2);
+ }
+ // complete with original exception
+ promise.completeExceptionally(exception);
+ });
+ } else {
+ log.info("kept ledger:{}", ledger.getId());
+ promise.complete(ledger.getId());
+ }
+ });
+
+ return promise;
+ }
+
+ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
+ LedgerHandle lh, Semaphore outstanding,
+ CompletableFuture<Void> promise) {
+ if (promise.isDone()) {
+ return;
+ }
+ CompletableFuture.runAsync(() -> {
+ if (reader.hasNext()) {
+ Message<T> message = reader.next();
+ mxBean.addCompactionReadOp(topic, message.size());
+ addToCompactedLedger(lh, message, topic, outstanding)
+ .whenComplete((res, exception2) -> {
+ outstanding.release();
+ if (exception2 != null) {
+ promise.completeExceptionally(exception2);
+ return;
+ }
+ });
+ phaseTwoLoop(topic, reader, lh, outstanding, promise);
Review Comment:
No. We don't want to wait for each `addToCompactedLedger` call to complete for high concurrency (this concurrency is limited by the semaphore, `outstanding`).
--
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 #18195: [improve][broker] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1017284346
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.client.impl;
+
+import io.netty.buffer.ByteBuf;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.MessageCrypto;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
+import org.apache.pulsar.common.api.EncryptionContext;
+import org.apache.pulsar.common.api.proto.CompressionType;
+import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.compression.CompressionCodecNone;
+import org.apache.pulsar.common.compression.CompressionCodecProvider;
+import org.apache.pulsar.common.protocol.Commands;
+
+/**
+ * A raw batch message container without producer. (Used for StrategicTwoPhaseCompactor)
+ *
+ * incoming single messages:
+ * (k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)
+ *
+ * batched into single batch message:
+ * [(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;
+ public RawBatchMessageContainerImpl(int maxNumMessagesInBatch) {
+ super();
+ compressionType = CompressionType.NONE;
+ compressor = new CompressionCodecNone();
+ if (maxNumMessagesInBatch > 0) {
+ this.maxNumMessagesInBatch = maxNumMessagesInBatch;
+ }
+ }
+ private ByteBuf encrypt(ByteBuf compressedPayload) {
+ if (msgCrypto == null) {
+ return compressedPayload;
+ }
+ int maxSize = msgCrypto.getMaxOutputSize(compressedPayload.readableBytes());
+ ByteBuf encryptedPayload = allocator.buffer(maxSize);
+ ByteBuffer targetBuffer = encryptedPayload.nioBuffer(0, maxSize);
+
+ try {
+ msgCrypto.encrypt(encryptionKeys, cryptoKeyReader, () -> messageMetadata,
+ compressedPayload.nioBuffer(), targetBuffer);
+ } catch (PulsarClientException e) {
+ encryptedPayload.release();
+ compressedPayload.release();
+ discard(e);
+ throw new RuntimeException("Failed to encrypt payload", e);
+ }
+ encryptedPayload.writerIndex(targetBuffer.remaining());
+ compressedPayload.release();
+ return encryptedPayload;
+ }
+
+ @Override
+ public ProducerImpl.OpSendMsg createOpSendMsg() {
+ throw new UnsupportedOperationException();
+ }
+ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
+ this.cryptoKeyReader = cryptoKeyReader;
+ }
+
+ public ByteBuf toByteBuf() {
Review Comment:
Ack.
I will add the comments when resolving comments from others.
--
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 #18195: [improve][broker] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#issuecomment-1306131631
# [Codecov](https://codecov.io/gh/apache/pulsar/pull/18195?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#18195](https://codecov.io/gh/apache/pulsar/pull/18195?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (b27bdfb) into [master](https://codecov.io/gh/apache/pulsar/commit/6c65ca0d8a80bfaaa4d5869e0cea485f5c94369b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (6c65ca0) will **decrease** coverage by `25.59%`.
> The diff coverage is `0.00%`.
[](https://codecov.io/gh/apache/pulsar/pull/18195?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #18195 +/- ##
=============================================
- Coverage 34.91% 9.32% -25.60%
+ Complexity 5707 1938 -3769
=============================================
Files 607 590 -17
Lines 53396 56469 +3073
Branches 5712 5870 +158
=============================================
- Hits 18644 5264 -13380
- Misses 32119 50616 +18497
+ Partials 2633 589 -2044
```
| Flag | Coverage Δ | |
|---|---|---|
| unittests | `9.32% <0.00%> (-25.60%)` | :arrow_down: |
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=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/pulsar/pull/18195?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...in/java/org/apache/pulsar/PulsarBrokerStarter.java](https://codecov.io/gh/apache/pulsar/pull/18195/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL1B1bHNhckJyb2tlclN0YXJ0ZXIuamF2YQ==) | `0.00% <ø> (ø)` | |
| [.../org/apache/pulsar/PulsarClusterMetadataSetup.java](https://codecov.io/gh/apache/pulsar/pull/18195/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL1B1bHNhckNsdXN0ZXJNZXRhZGF0YVNldHVwLmphdmE=) | `0.00% <ø> (ø)` | |
| [...g/apache/pulsar/PulsarClusterMetadataTeardown.java](https://codecov.io/gh/apache/pulsar/pull/18195/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL1B1bHNhckNsdXN0ZXJNZXRhZGF0YVRlYXJkb3duLmphdmE=) | `0.00% <ø> (ø)` | |
| [...org/apache/pulsar/PulsarInitialNamespaceSetup.java](https://codecov.io/gh/apache/pulsar/pull/18195/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL1B1bHNhckluaXRpYWxOYW1lc3BhY2VTZXR1cC5qYXZh) | `0.00% <ø> (ø)` | |
| [.../main/java/org/apache/pulsar/PulsarStandalone.java](https://codecov.io/gh/apache/pulsar/pull/18195/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL1B1bHNhclN0YW5kYWxvbmUuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...ava/org/apache/pulsar/PulsarStandaloneBuilder.java](https://codecov.io/gh/apache/pulsar/pull/18195/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL1B1bHNhclN0YW5kYWxvbmVCdWlsZGVyLmphdmE=) | `0.00% <ø> (ø)` | |
| [...ava/org/apache/pulsar/broker/admin/v2/Brokers.java](https://codecov.io/gh/apache/pulsar/pull/18195/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi92Mi9Ccm9rZXJzLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...ava/org/apache/pulsar/broker/admin/v2/Tenants.java](https://codecov.io/gh/apache/pulsar/pull/18195/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi92Mi9UZW5hbnRzLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...va/org/apache/pulsar/broker/admin/v2/Clusters.java](https://codecov.io/gh/apache/pulsar/pull/18195/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi92Mi9DbHVzdGVycy5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...ache/pulsar/broker/service/PublishRateLimiter.java](https://codecov.io/gh/apache/pulsar/pull/18195/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL1B1Ymxpc2hSYXRlTGltaXRlci5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| ... and [679 more](https://codecov.io/gh/apache/pulsar/pull/18195/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
--
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 #18195: [improve][broker] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
Demogorgon314 commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1015050223
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -0,0 +1,433 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.impl.BatchMessageIdImpl;
+import org.apache.pulsar.client.impl.CompactionReaderImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.RawBatchMessageContainerImpl;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects valid message(defined in the TopicCompactionStrategy.isValid())
+ * for each key in the topic. Then, the second pass writes these values
+ * to a ledger.
+ *
+ * <p>As the first pass caches the entire message(not just offset) for each key into a map,
+ * this compaction could be memory intensive if the message payload is large.
+ */
+public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
+ private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
+ private static final int MAX_OUTSTANDING = 500;
+ private final Duration phaseOneLoopReadTimeout;
+ private final RawBatchMessageContainerImpl batchMessageContainer;
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler,
+ int maxNumMessagesInBatch) {
+ super(conf, pulsar, bk, scheduler);
+ batchMessageContainer = new RawBatchMessageContainerImpl(maxNumMessagesInBatch);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+ }
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this(conf, pulsar, bk, scheduler, -1);
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy) {
+ return compact(topic, strategy, null);
+ }
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy,
+ CryptoKeyReader cryptoKeyReader) {
+ CompletableFuture<Consumer<T>> consumerFuture = new CompletableFuture<>();
+ if (cryptoKeyReader != null) {
+ batchMessageContainer.setCryptoKeyReader(cryptoKeyReader);
+ }
+ CompactionReaderImpl reader = CompactionReaderImpl.create(
+ (PulsarClientImpl) pulsar, strategy.getSchema(), topic, consumerFuture, cryptoKeyReader);
+
+ return consumerFuture.thenComposeAsync(__ -> compactAndCloseReader(reader, strategy), scheduler);
+ }
+
+ <T> CompletableFuture<Long> doCompaction(Reader<T> reader, TopicCompactionStrategy strategy) {
+
+ if (!(reader instanceof CompactionReaderImpl<T>)) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("reader has to be DelayedAckReaderImpl"));
+ }
+ return reader.hasMessageAvailableAsync()
+ .thenCompose(available -> {
+ if (available) {
+ return phaseOne(reader, strategy)
+ .thenCompose((result) -> phaseTwo(result, reader, bk));
+ } else {
+ log.info("Skip compaction of the empty topic {}", reader.getTopic());
+ return CompletableFuture.completedFuture(-1L);
+ }
+ });
+ }
+
+ <T> CompletableFuture<Long> compactAndCloseReader(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ mxBean.addCompactionStartOp(reader.getTopic());
+ doCompaction(reader, strategy).whenComplete(
+ (ledgerId, exception) -> {
+ log.info("Completed doCompaction ledgerId:{}", ledgerId);
+ reader.closeAsync().whenComplete((v, exception2) -> {
+ if (exception2 != null) {
+ log.warn("Error closing reader handle {}, ignoring", reader, exception2);
+ }
+ if (exception != null) {
+ // complete with original exception
+ mxBean.addCompactionEndOp(reader.getTopic(), false);
+ promise.completeExceptionally(exception);
+ } else {
+ mxBean.addCompactionEndOp(reader.getTopic(), true);
+ promise.complete(ledgerId);
+ }
+ });
+ });
+ return promise;
+ }
+
+ private <T> boolean doCompactMessage(
+ Message<T> msg, PhaseOneResult<T> result, TopicCompactionStrategy<T> strategy) {
+ Map<String, Message<T>> cache = result.cache;
+ String key = msg.getKey();
+
+ if (key == null) {
+ msg.release();
+ return true;
+ }
+ T val = msg.getValue();
+ Message<T> prev = cache.get(key);
+ T prevVal = prev == null ? null : prev.getValue();
+
+ if (!strategy.shouldKeepLeft(prevVal, val)) {
+ if (val != null && msg.size() > 0) {
+ cache.remove(key); // to reorder
+ cache.put(key, msg);
+ } else {
+ cache.remove(key);
+ msg.release();
+ }
+
+ if (prev != null) {
+ prev.release();
+ }
+
+ result.validCompactionCount.incrementAndGet();
+ return true;
+ } else {
+ msg.release();
+ result.invalidCompactionCount.incrementAndGet();
+ return false;
+ }
+
+ }
+
+ private static class PhaseOneResult<T> {
+ MessageId firstId;
+ MessageId lastId; // last read messageId
+ Map<String, Message<T>> cache;
+
+ AtomicInteger invalidCompactionCount;
+
+ AtomicInteger validCompactionCount;
+
+ AtomicInteger numReadMessages;
+
+ String topic;
+
+ PhaseOneResult(String topic) {
+ this.topic = topic;
+ cache = new LinkedHashMap<>();
+ invalidCompactionCount = new AtomicInteger();
+ validCompactionCount = new AtomicInteger();
+ numReadMessages = new AtomicInteger();
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "{Topic:%s, firstId:%s, lastId:%s, cache.size:%d, "
+ + "invalidCompactionCount:%d, validCompactionCount:%d, numReadMessages:%d}",
+ topic,
+ firstId != null ? firstId.toString() : "",
+ lastId != null ? lastId.toString() : "",
+ cache.size(),
+ invalidCompactionCount.get(),
+ validCompactionCount.get(),
+ numReadMessages.get());
+ }
+ }
+
+
+ private <T> CompletableFuture<PhaseOneResult> phaseOne(Reader<T> reader, TopicCompactionStrategy strategy) {
+ CompletableFuture<PhaseOneResult> promise = new CompletableFuture<>();
+ PhaseOneResult<T> result = new PhaseOneResult(reader.getTopic());
+
+ ((CompactionReaderImpl<T>) reader).getLastMessageIdAsync()
+ .thenAccept(lastMessageId -> {
+ log.info("Commencing phase one of compaction for {}, reading to {}",
+ reader.getTopic(), lastMessageId);
+ result.lastId = copyMessageId(lastMessageId);
+ phaseOneLoop(reader, promise, result, strategy);
+ }).exceptionally(ex -> {
+ promise.completeExceptionally(ex);
+ return null;
+ });
+
+ return promise;
+
+ }
+
+ private static MessageId copyMessageId(MessageId msgId) {
+ if (msgId instanceof BatchMessageIdImpl) {
+ BatchMessageIdImpl tempId = (BatchMessageIdImpl) msgId;
+ return new BatchMessageIdImpl(tempId);
+ } else if (msgId instanceof MessageIdImpl) {
+ MessageIdImpl tempId = (MessageIdImpl) msgId;
+ return new MessageIdImpl(tempId.getLedgerId(), tempId.getEntryId(),
+ tempId.getPartitionIndex());
+ } else {
+ throw new IllegalStateException("Unknown lastMessageId type");
+ }
+ }
+
+ private <T> void phaseOneLoop(Reader<T> reader, CompletableFuture<PhaseOneResult> promise,
+ PhaseOneResult<T> result, TopicCompactionStrategy<T> strategy) {
+
+ if (promise.isDone()) {
+ return;
+ }
+
+ CompletableFuture<Message<T>> future = reader.readNextAsync();
+ FutureUtil.addTimeoutHandling(future,
+ phaseOneLoopReadTimeout, scheduler,
+ () -> FutureUtil.createTimeoutException("Timeout", getClass(),
+ "phaseOneLoop(...)"));
+
+ future.thenAcceptAsync(msg -> {
+
+ MessageId id = msg.getMessageId();
+ boolean completed = false;
+ if (result.lastId.compareTo(id) == 0) {
+ completed = true;
+ }
+
+ result.numReadMessages.incrementAndGet();
+ mxBean.addCompactionReadOp(reader.getTopic(), msg.size());
+ if (doCompactMessage(msg, result, strategy)) {
+ mxBean.addCompactionRemovedEvent(reader.getTopic());
+ }
+ //set ids in the result
+ if (result.firstId == null) {
+ result.firstId = copyMessageId(id);
+ log.info("Resetting cursor to firstId:{}", result.firstId);
+ try {
+ reader.seek(result.firstId);
+ } catch (PulsarClientException e) {
+ throw new RuntimeException("Failed to reset the cursor to firstId:" + result.firstId, e);
+ }
+ }
+ if (completed) {
+ promise.complete(result);
+ } else {
+ phaseOneLoop(reader, promise, result, strategy);
+ }
+
+ }, scheduler).exceptionally(ex -> {
+ promise.completeExceptionally(ex);
+ return null;
+ });
+
+ }
+
+ private <T> CompletableFuture<Long> phaseTwo(PhaseOneResult<T> phaseOneResult, Reader<T> reader, BookKeeper bk) {
+ log.info("Completed phase one. Result:{}. ", phaseOneResult);
+ Map<String, byte[]> metadata =
+ LedgerMetadataUtils.buildMetadataForCompactedLedger(
+ phaseOneResult.topic, phaseOneResult.lastId.toByteArray());
+ return createLedger(bk, metadata)
+ .thenCompose((ledger) -> {
+ log.info(
+ "Commencing phase two of compaction for {}, from {} to {}, compacting {} keys to ledger {}",
+ phaseOneResult.topic, phaseOneResult.firstId, phaseOneResult.lastId,
+ phaseOneResult.cache.size(), ledger.getId());
+ return runPhaseTwo(phaseOneResult, reader, ledger, bk);
+ });
+ }
+
+ private <T> CompletableFuture<Long> runPhaseTwo(
+ PhaseOneResult<T> phaseOneResult, Reader<T> reader, LedgerHandle ledger, BookKeeper bk) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ Semaphore outstanding = new Semaphore(MAX_OUTSTANDING);
+ CompletableFuture<Void> loopPromise = new CompletableFuture<>();
+ phaseTwoLoop(phaseOneResult.topic, phaseOneResult.cache.values().iterator(), ledger,
+ outstanding, loopPromise);
+ loopPromise.thenCompose((v) -> {
+ log.info("Flushing batch container numMessagesInBatch:{}",
+ batchMessageContainer.getNumMessagesInBatch());
+ return addToCompactedLedger(ledger, null, reader.getTopic(), outstanding)
+ .whenComplete((res, exception2) -> {
+ if (exception2 != null) {
+ promise.completeExceptionally(exception2);
+ return;
+ }
+ });
+ })
+ .thenCompose(v -> {
+ log.info("Acking ledger id {}", phaseOneResult.firstId);
+ return ((CompactionReaderImpl<T>) reader)
+ .acknowledgeCumulativeAsync(
+ phaseOneResult.lastId, Map.of(COMPACTED_TOPIC_LEDGER_PROPERTY,
+ ledger.getId()));
+ })
+ .thenCompose((v) -> closeLedger(ledger))
+ .whenComplete((v, exception) -> {
+ if (exception != null) {
+ deleteLedger(bk, ledger).whenComplete((res2, exception2) -> {
+ if (exception2 != null) {
+ log.error("Cleanup of ledger {} for failed", ledger, exception2);
+ }
+ // complete with original exception
+ promise.completeExceptionally(exception);
+ });
+ } else {
+ log.info("kept ledger:{}", ledger.getId());
+ promise.complete(ledger.getId());
+ }
+ });
+
+ return promise;
+ }
+
+ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
+ LedgerHandle lh, Semaphore outstanding,
+ CompletableFuture<Void> promise) {
+ if (promise.isDone()) {
+ return;
+ }
+ CompletableFuture.runAsync(() -> {
+ if (reader.hasNext()) {
+ Message<T> message = reader.next();
+ mxBean.addCompactionReadOp(topic, message.size());
+ addToCompactedLedger(lh, message, topic, outstanding)
+ .whenComplete((res, exception2) -> {
+ outstanding.release();
+ if (exception2 != null) {
+ promise.completeExceptionally(exception2);
+ return;
+ }
+ });
+ phaseTwoLoop(topic, reader, lh, outstanding, promise);
Review Comment:
```suggestion
addToCompactedLedger(lh, message, topic, outstanding)
.whenComplete((res, exception2) -> {
outstanding.release();
if (exception2 != null) {
promise.completeExceptionally(exception2);
return;
}
phaseTwoLoop(topic, reader, lh, outstanding, promise);
});
```
If I understand correctly, the logic should be like this, 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] codelipenghui commented on a diff in pull request #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1030343532
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewConfigurationData.java:
##########
@@ -29,13 +29,15 @@ public class TableViewConfigurationData implements Serializable, Cloneable {
private String topicName = null;
private long autoUpdatePartitionsSeconds = 60;
+ private String topicCompactionStrategy;
Review Comment:
```suggestion
private String topicCompactionStrategyClassName;
```
And why not use `TopicCompactionStrategy topicCompactionStrategy` directly? If we don't want to expose it as a public API, we can just ignore the serialization of the `topicCompactionStrategy`
##########
pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicCompactionStrategy.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.common.topics;
+
+import org.apache.pulsar.client.api.Schema;
+
+/**
+ * Defines a custom strategy to compact messages in a topic.
+ * This strategy can be passed to Topic Compactor and Table View to compact messages in a custom way.
+ *
+ * Examples:
+ *
+ * TopicCompactionStrategy strategy = new MyTopicCompactionStrategy();
+ *
+ * // Run topic compaction by the compaction strategy.
+ * // While compacting messages for each key,
+ * // it will choose messages only if TopicCompactionStrategy.shouldKeepLeft(prev, cur) returns false.
+ * StrategicTwoPhaseCompactor compactor = new StrategicTwoPhaseCompactor(...);
+ * compactor.compact(topic, strategy);
+ *
+ * // Run table view by the compaction strategy.
+ * // While updating messages in the table view <key,value> map,
+ * // it will choose messages only if TopicCompactionStrategy.shouldKeepLeft(prev, cur) returns false.
+ * TableView tableView = pulsar.getClient().newTableViewBuilder(strategy.getSchema())
+ * .topic(topic)
+ * .loadConf(Map.of(
+ * "topicCompactionStrategy", strategy.getClass().getCanonicalName()))
+ * .create();
+ */
+public interface TopicCompactionStrategy<T> {
+
+ /**
+ * Returns the schema object for this strategy.
+ * @return
+ */
+ Schema<T> getSchema();
Review Comment:
Why the compaction strategy should provide the schema? IMO, the scope of the `TopicCompactionStrategy` is to return an accepted value from 2 values of a key.
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -0,0 +1,434 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.impl.BatchMessageIdImpl;
+import org.apache.pulsar.client.impl.CompactionReaderImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.RawBatchMessageContainerImpl;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Compaction will go through the topic in two passes. The first pass
+ * selects valid message(defined in the TopicCompactionStrategy.isValid())
+ * for each key in the topic. Then, the second pass writes these values
+ * to a ledger.
+ *
+ * <p>As the first pass caches the entire message(not just offset) for each key into a map,
+ * this compaction could be memory intensive if the message payload is large.
+ */
+public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
+ private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
+ private static final int MAX_OUTSTANDING = 500;
+ private final Duration phaseOneLoopReadTimeout;
+ private final RawBatchMessageContainerImpl batchMessageContainer;
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler,
+ int maxNumMessagesInBatch) {
+ super(conf, pulsar, bk, scheduler);
+ batchMessageContainer = new RawBatchMessageContainerImpl(maxNumMessagesInBatch);
+ phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+ }
+
+ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
+ PulsarClient pulsar,
+ BookKeeper bk,
+ ScheduledExecutorService scheduler) {
+ this(conf, pulsar, bk, scheduler, -1);
+ }
+
+ public CompletableFuture<Long> compact(String topic) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy) {
+ return compact(topic, strategy, null);
+ }
+
+ public <T> CompletableFuture<Long> compact(String topic,
+ TopicCompactionStrategy<T> strategy,
+ CryptoKeyReader cryptoKeyReader) {
+ CompletableFuture<Consumer<T>> consumerFuture = new CompletableFuture<>();
+ if (cryptoKeyReader != null) {
+ batchMessageContainer.setCryptoKeyReader(cryptoKeyReader);
+ }
+ CompactionReaderImpl reader = CompactionReaderImpl.create(
+ (PulsarClientImpl) pulsar, strategy.getSchema(), topic, consumerFuture, cryptoKeyReader);
+
+ return consumerFuture.thenComposeAsync(__ -> compactAndCloseReader(reader, strategy), scheduler);
+ }
+
+ <T> CompletableFuture<Long> doCompaction(Reader<T> reader, TopicCompactionStrategy strategy) {
+
+ if (!(reader instanceof CompactionReaderImpl<T>)) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("reader has to be DelayedAckReaderImpl"));
+ }
+ return reader.hasMessageAvailableAsync()
Review Comment:
We will introduce data deserialization if don't use RawReader here. The compaction is running inside the broker. The data deserialization will push more strain on Garbage Collection.
##########
pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicCompactionStrategy.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.common.topics;
+
+import org.apache.pulsar.client.api.Schema;
+
+/**
+ * Defines a custom strategy to compact messages in a topic.
+ * This strategy can be passed to Topic Compactor and Table View to compact messages in a custom way.
+ *
+ * Examples:
+ *
+ * TopicCompactionStrategy strategy = new MyTopicCompactionStrategy();
+ *
+ * // Run topic compaction by the compaction strategy.
+ * // While compacting messages for each key,
+ * // it will choose messages only if TopicCompactionStrategy.shouldKeepLeft(prev, cur) returns false.
+ * StrategicTwoPhaseCompactor compactor = new StrategicTwoPhaseCompactor(...);
+ * compactor.compact(topic, strategy);
+ *
+ * // Run table view by the compaction strategy.
+ * // While updating messages in the table view <key,value> map,
+ * // it will choose messages only if TopicCompactionStrategy.shouldKeepLeft(prev, cur) returns false.
+ * TableView tableView = pulsar.getClient().newTableViewBuilder(strategy.getSchema())
+ * .topic(topic)
+ * .loadConf(Map.of(
+ * "topicCompactionStrategy", strategy.getClass().getCanonicalName()))
+ * .create();
+ */
+public interface TopicCompactionStrategy<T> {
+
+ /**
+ * Returns the schema object for this strategy.
+ * @return
+ */
+ Schema<T> getSchema();
+ /**
+ * Tests if the compaction needs to keep the left(previous message)
+ * compared to the right(current message) for the same key.
+ *
+ * @param prev previous message value
+ * @param cur current message value
+ * @return True if it needs to keep the previous message and ignore the current message. Otherwise, False.
+ */
+ boolean shouldKeepLeft(T prev, T cur);
Review Comment:
Can we just use `T getCompactedValue(T v1, T v2)`?
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewConfigurationData.java:
##########
@@ -29,13 +29,15 @@ public class TableViewConfigurationData implements Serializable, Cloneable {
private String topicName = null;
private long autoUpdatePartitionsSeconds = 60;
+ private String topicCompactionStrategy;
Review Comment:
How will the broker know which strategy should use? Or it just happened on the client side. For the broker side, we need to set the strategy by setting topic policies.
--
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 #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1033147317
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewConfigurationData.java:
##########
@@ -29,13 +29,15 @@ public class TableViewConfigurationData implements Serializable, Cloneable {
private String topicName = null;
private long autoUpdatePartitionsSeconds = 60;
+ private String topicCompactionStrategy;
Review Comment:
> How will the broker know which strategy should use? Or it just happened on the client side. For the broker side, we need to set the strategy by setting topic policies.
client can configure any matching topic compaction strategy if the user knows the compaction class name.
Right, on the broker side, ideally we can eventually set the compaction class name in the topic policy.
But for pip192 i think we could start with a simple hashmap (topicname-> compact strategy) for the new load balancer's internal system topic to use the compaction strategy.
--
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] shibd commented on a diff in pull request #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
shibd commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1028176302
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.client.impl;
+
+import io.netty.buffer.ByteBuf;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.MessageCrypto;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
+import org.apache.pulsar.common.api.EncryptionContext;
+import org.apache.pulsar.common.api.proto.CompressionType;
+import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.compression.CompressionCodecNone;
+import org.apache.pulsar.common.compression.CompressionCodecProvider;
+import org.apache.pulsar.common.protocol.Commands;
+
+/**
+ * A raw batch message container without producer. (Used for StrategicTwoPhaseCompactor)
+ *
+ * incoming single messages:
+ * (k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)
+ *
+ * batched into single batch message:
+ * [(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]
+ */
+public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
Review Comment:
It seems that this is just a need for a bulk message container based on the maximum number of batches, and I feel that extending the `BatchMessageContainerImpl` will become too heavy(There are various producer-related properties and operations in the BatchMessageContainer implementation), and it also adds to the complexity of the original BatchMessageContainer.
Maybe a simple batch container could be re-implemented. What do you think?
--
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] shibd commented on a diff in pull request #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
shibd commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1028740540
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.client.impl;
+
+import io.netty.buffer.ByteBuf;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.MessageCrypto;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
+import org.apache.pulsar.common.api.EncryptionContext;
+import org.apache.pulsar.common.api.proto.CompressionType;
+import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.compression.CompressionCodecNone;
+import org.apache.pulsar.common.compression.CompressionCodecProvider;
+import org.apache.pulsar.common.protocol.Commands;
+
+/**
+ * A raw batch message container without producer. (Used for StrategicTwoPhaseCompactor)
+ *
+ * incoming single messages:
+ * (k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)
+ *
+ * batched into single batch message:
+ * [(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]
+ */
+public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
Review Comment:
My main concern is that developers will miss this later when they modify it. Using mock producer is transfer behavior to `RawBatchMessageContainerImpl `.
I revisited the code, and the current implementation is not quite capable of the mock producer.
like this:
https://github.com/apache/pulsar/blob/4faed760376d6c5bec8803f97c4ed2452664ac69/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java#L116-L117
https://github.com/apache/pulsar/blob/4faed760376d6c5bec8803f97c4ed2452664ac69/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java#L136-L138
BWT: In the original implementation, `BatchContainer` coupled with the producer I didn't think was very good. We should make `BatchContiner` independent, Leave things like `releaseMemory` to the producer. (This is off topic and does not affect this PR)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: 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 #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1033158734
##########
pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicCompactionStrategy.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.common.topics;
+
+import org.apache.pulsar.client.api.Schema;
+
+/**
+ * Defines a custom strategy to compact messages in a topic.
+ * This strategy can be passed to Topic Compactor and Table View to compact messages in a custom way.
+ *
+ * Examples:
+ *
+ * TopicCompactionStrategy strategy = new MyTopicCompactionStrategy();
+ *
+ * // Run topic compaction by the compaction strategy.
+ * // While compacting messages for each key,
+ * // it will choose messages only if TopicCompactionStrategy.shouldKeepLeft(prev, cur) returns false.
+ * StrategicTwoPhaseCompactor compactor = new StrategicTwoPhaseCompactor(...);
+ * compactor.compact(topic, strategy);
+ *
+ * // Run table view by the compaction strategy.
+ * // While updating messages in the table view <key,value> map,
+ * // it will choose messages only if TopicCompactionStrategy.shouldKeepLeft(prev, cur) returns false.
+ * TableView tableView = pulsar.getClient().newTableViewBuilder(strategy.getSchema())
+ * .topic(topic)
+ * .loadConf(Map.of(
+ * "topicCompactionStrategy", strategy.getClass().getCanonicalName()))
+ * .create();
+ */
+public interface TopicCompactionStrategy<T> {
+
+ /**
+ * Returns the schema object for this strategy.
+ * @return
+ */
+ Schema<T> getSchema();
Review Comment:
This is to specify the schema required for this strategy.
Otherwise, we need to infer that the type of this strategy is the schema.
--
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 #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1033143208
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewConfigurationData.java:
##########
@@ -29,13 +29,15 @@ public class TableViewConfigurationData implements Serializable, Cloneable {
private String topicName = null;
private long autoUpdatePartitionsSeconds = 60;
+ private String topicCompactionStrategy;
Review Comment:
I couldn't find a strong benefit to pass 'TopicCompactionStrategy' obj in this config.
I thought this config needs to be light-weight 'configs' instead of passing heavy objects.
--
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 #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1053790307
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.client.impl;
+
+import io.netty.buffer.ByteBuf;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.MessageCrypto;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
+import org.apache.pulsar.common.api.EncryptionContext;
+import org.apache.pulsar.common.api.proto.CompressionType;
+import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.compression.CompressionCodecNone;
+import org.apache.pulsar.common.compression.CompressionCodecProvider;
+import org.apache.pulsar.common.protocol.Commands;
+
+/**
+ * A raw batch message container without producer. (Used for StrategicTwoPhaseCompactor)
+ *
+ * incoming single messages:
+ * (k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)
+ *
+ * batched into single batch message:
+ * [(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;
+ public RawBatchMessageContainerImpl(int maxNumMessagesInBatch) {
+ super();
+ compressionType = CompressionType.NONE;
+ compressor = new CompressionCodecNone();
+ if (maxNumMessagesInBatch > 0) {
+ this.maxNumMessagesInBatch = maxNumMessagesInBatch;
+ }
+ }
+ private ByteBuf encrypt(ByteBuf compressedPayload) {
+ if (msgCrypto == null) {
+ return compressedPayload;
+ }
+ int maxSize = msgCrypto.getMaxOutputSize(compressedPayload.readableBytes());
+ ByteBuf encryptedPayload = allocator.buffer(maxSize);
+ ByteBuffer targetBuffer = encryptedPayload.nioBuffer(0, maxSize);
+
+ try {
+ msgCrypto.encrypt(encryptionKeys, cryptoKeyReader, () -> messageMetadata,
+ compressedPayload.nioBuffer(), targetBuffer);
+ } catch (PulsarClientException e) {
+ encryptedPayload.release();
+ compressedPayload.release();
+ discard(e);
+ throw new RuntimeException("Failed to encrypt payload", e);
+ }
+ encryptedPayload.writerIndex(targetBuffer.remaining());
+ compressedPayload.release();
+ return encryptedPayload;
+ }
+
+ @Override
+ public ProducerImpl.OpSendMsg createOpSendMsg() {
+ throw new UnsupportedOperationException();
+ }
+ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
+ this.cryptoKeyReader = cryptoKeyReader;
+ }
+
+ public ByteBuf toByteBuf() {
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] heesung-sn commented on a diff in pull request #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1053791587
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewConfigurationData.java:
##########
@@ -29,13 +29,15 @@ public class TableViewConfigurationData implements Serializable, Cloneable {
private String topicName = null;
private long autoUpdatePartitionsSeconds = 60;
+ private String topicCompactionStrategy;
Review Comment:
updated this name to `topicCompactionStrategyClassName`
--
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 pull request #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#issuecomment-1359962458
Rebased this pip-215 branch on top of the master.
--
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 #18195: [improve][broker] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1015801846
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.client.impl;
+
+import io.netty.buffer.ByteBuf;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.MessageCrypto;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
+import org.apache.pulsar.common.api.EncryptionContext;
+import org.apache.pulsar.common.api.proto.CompressionType;
+import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.compression.CompressionCodecNone;
+import org.apache.pulsar.common.compression.CompressionCodecProvider;
+import org.apache.pulsar.common.protocol.Commands;
+
+/**
+ * A raw batch message container without producer. (Used for StrategicTwoPhaseCompactor)
+ *
+ * incoming single messages:
+ * (k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)
+ *
+ * batched into single batch message:
+ * [(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;
+ public RawBatchMessageContainerImpl(int maxNumMessagesInBatch) {
+ super();
+ compressionType = CompressionType.NONE;
+ compressor = new CompressionCodecNone();
+ if (maxNumMessagesInBatch > 0) {
+ this.maxNumMessagesInBatch = maxNumMessagesInBatch;
+ }
+ }
+ private ByteBuf encrypt(ByteBuf compressedPayload) {
+ if (msgCrypto == null) {
+ return compressedPayload;
+ }
+ int maxSize = msgCrypto.getMaxOutputSize(compressedPayload.readableBytes());
+ ByteBuf encryptedPayload = allocator.buffer(maxSize);
+ ByteBuffer targetBuffer = encryptedPayload.nioBuffer(0, maxSize);
+
+ try {
+ msgCrypto.encrypt(encryptionKeys, cryptoKeyReader, () -> messageMetadata,
+ compressedPayload.nioBuffer(), targetBuffer);
+ } catch (PulsarClientException e) {
+ throw new RuntimeException("Failed to encrypt payload", e);
+ }
+ encryptedPayload.writerIndex(targetBuffer.remaining());
+ compressedPayload.release();
+ return encryptedPayload;
+ }
+
+ @Override
+ public ProducerImpl.OpSendMsg createOpSendMsg() {
+ throw new UnsupportedOperationException();
+ }
+ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
+ this.cryptoKeyReader = cryptoKeyReader;
+ }
+
+ public ByteBuf toByteBuf() {
+ if (numMessagesInBatch > 1) {
+ messageMetadata.setNumMessagesInBatch(numMessagesInBatch);
+ messageMetadata.setSequenceId(lowestSequenceId);
+ messageMetadata.setHighestSequenceId(highestSequenceId);
+ }
+ MessageImpl lastMessage = messages.get(messages.size() - 1);
+ MessageIdImpl lastMessageId = (MessageIdImpl) lastMessage.getMessageId();
+ MessageMetadata lastMessageMetadata = lastMessage.getMessageBuilder();
+
+ this.compressionType = lastMessageMetadata.getCompression();
+ this.compressor = CompressionCodecProvider.getCompressionCodec(lastMessageMetadata.getCompression());
+
+ if (!lastMessage.getEncryptionCtx().isEmpty()) {
+ EncryptionContext encryptionContext = (EncryptionContext) lastMessage.getEncryptionCtx().get();
+
+ if (cryptoKeyReader == null) {
+ throw new IllegalStateException("Messages are encrypted but no cryptoKeyReader is provided.");
Review Comment:
Yes, I think we better clean the container from these exceptional cases.
I added `discard()` from those exceptions. (discard() internally calls 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] heesung-sn commented on a diff in pull request #18195: [improve][broker] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
heesung-sn commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1015801846
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.client.impl;
+
+import io.netty.buffer.ByteBuf;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.MessageCrypto;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
+import org.apache.pulsar.common.api.EncryptionContext;
+import org.apache.pulsar.common.api.proto.CompressionType;
+import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.compression.CompressionCodecNone;
+import org.apache.pulsar.common.compression.CompressionCodecProvider;
+import org.apache.pulsar.common.protocol.Commands;
+
+/**
+ * A raw batch message container without producer. (Used for StrategicTwoPhaseCompactor)
+ *
+ * incoming single messages:
+ * (k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)
+ *
+ * batched into single batch message:
+ * [(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;
+ public RawBatchMessageContainerImpl(int maxNumMessagesInBatch) {
+ super();
+ compressionType = CompressionType.NONE;
+ compressor = new CompressionCodecNone();
+ if (maxNumMessagesInBatch > 0) {
+ this.maxNumMessagesInBatch = maxNumMessagesInBatch;
+ }
+ }
+ private ByteBuf encrypt(ByteBuf compressedPayload) {
+ if (msgCrypto == null) {
+ return compressedPayload;
+ }
+ int maxSize = msgCrypto.getMaxOutputSize(compressedPayload.readableBytes());
+ ByteBuf encryptedPayload = allocator.buffer(maxSize);
+ ByteBuffer targetBuffer = encryptedPayload.nioBuffer(0, maxSize);
+
+ try {
+ msgCrypto.encrypt(encryptionKeys, cryptoKeyReader, () -> messageMetadata,
+ compressedPayload.nioBuffer(), targetBuffer);
+ } catch (PulsarClientException e) {
+ throw new RuntimeException("Failed to encrypt payload", e);
+ }
+ encryptedPayload.writerIndex(targetBuffer.remaining());
+ compressedPayload.release();
+ return encryptedPayload;
+ }
+
+ @Override
+ public ProducerImpl.OpSendMsg createOpSendMsg() {
+ throw new UnsupportedOperationException();
+ }
+ public void setCryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
+ this.cryptoKeyReader = cryptoKeyReader;
+ }
+
+ public ByteBuf toByteBuf() {
+ if (numMessagesInBatch > 1) {
+ messageMetadata.setNumMessagesInBatch(numMessagesInBatch);
+ messageMetadata.setSequenceId(lowestSequenceId);
+ messageMetadata.setHighestSequenceId(highestSequenceId);
+ }
+ MessageImpl lastMessage = messages.get(messages.size() - 1);
+ MessageIdImpl lastMessageId = (MessageIdImpl) lastMessage.getMessageId();
+ MessageMetadata lastMessageMetadata = lastMessage.getMessageBuilder();
+
+ this.compressionType = lastMessageMetadata.getCompression();
+ this.compressor = CompressionCodecProvider.getCompressionCodec(lastMessageMetadata.getCompression());
+
+ if (!lastMessage.getEncryptionCtx().isEmpty()) {
+ EncryptionContext encryptionContext = (EncryptionContext) lastMessage.getEncryptionCtx().get();
+
+ if (cryptoKeyReader == null) {
+ throw new IllegalStateException("Messages are encrypted but no cryptoKeyReader is provided.");
Review Comment:
Yes, I think we better clean the container from these exceptional cases.
I added `discard()` from those exceptions. (`discard()` internally calls `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] shibd commented on a diff in pull request #18195: [improve][broker][client] PIP-215 Added TopicCompactionStrategy for StrategicTwoPhaseCompactor and TableView.
Posted by GitBox <gi...@apache.org>.
shibd commented on code in PR #18195:
URL: https://github.com/apache/pulsar/pull/18195#discussion_r1028646491
##########
pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.client.impl;
+
+import io.netty.buffer.ByteBuf;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.MessageCrypto;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
+import org.apache.pulsar.common.api.EncryptionContext;
+import org.apache.pulsar.common.api.proto.CompressionType;
+import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.compression.CompressionCodecNone;
+import org.apache.pulsar.common.compression.CompressionCodecProvider;
+import org.apache.pulsar.common.protocol.Commands;
+
+/**
+ * A raw batch message container without producer. (Used for StrategicTwoPhaseCompactor)
+ *
+ * incoming single messages:
+ * (k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)
+ *
+ * batched into single batch message:
+ * [(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]
+ */
+public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
Review Comment:
oh, sorry, I miss some code.
> We could pass a mock producer if adding the producer null-check is not desirable.
I'm leaning towards this.
other, I wonder why batch processing is needed here? Is it because it is known that there is a performance bottleneck 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