You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/10/30 03:44:15 UTC
(pulsar) branch branch-3.1 updated: [fix][broker] Ignore individual acknowledgment for CompactorSubscription when an entry has been filtered. (#21434)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.1 by this push:
new 7a2e3f67b55 [fix][broker] Ignore individual acknowledgment for CompactorSubscription when an entry has been filtered. (#21434)
7a2e3f67b55 is described below
commit 7a2e3f67b552cf3436ca7b62bebe2a50f8235d66
Author: Cong Zhao <zh...@apache.org>
AuthorDate: Wed Oct 25 21:27:37 2023 +0800
[fix][broker] Ignore individual acknowledgment for CompactorSubscription when an entry has been filtered. (#21434)
---
.../broker/service/AbstractBaseDispatcher.java | 16 +++---
.../broker/service/plugin/FilterEntryTest.java | 61 +++++++++++++++++++++-
2 files changed, 68 insertions(+), 9 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
index eb8b0151395..b36389ab2da 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
@@ -176,14 +176,16 @@ public abstract class AbstractBaseDispatcher extends EntryFilterSupport implemen
if (Markers.isTxnMarker(msgMetadata)) {
// because consumer can receive message is smaller than maxReadPosition,
// so this marker is useless for this subscription
- individualAcknowledgeMessageIfNeeded(entry.getPosition(), Collections.emptyMap());
+ individualAcknowledgeMessageIfNeeded(Collections.singletonList(entry.getPosition()),
+ Collections.emptyMap());
entries.set(i, null);
entry.release();
continue;
} else if (((PersistentTopic) subscription.getTopic())
.isTxnAborted(new TxnID(msgMetadata.getTxnidMostBits(), msgMetadata.getTxnidLeastBits()),
(PositionImpl) entry.getPosition())) {
- individualAcknowledgeMessageIfNeeded(entry.getPosition(), Collections.emptyMap());
+ individualAcknowledgeMessageIfNeeded(Collections.singletonList(entry.getPosition()),
+ Collections.emptyMap());
entries.set(i, null);
entry.release();
continue;
@@ -200,7 +202,8 @@ public abstract class AbstractBaseDispatcher extends EntryFilterSupport implemen
entries.set(i, null);
entry.release();
- individualAcknowledgeMessageIfNeeded(pos, Collections.emptyMap());
+ individualAcknowledgeMessageIfNeeded(Collections.singletonList(pos),
+ Collections.emptyMap());
continue;
} else if (trackDelayedDelivery(entry.getLedgerId(), entry.getEntryId(), msgMetadata)) {
// The message is marked for delayed delivery. Ignore for now.
@@ -271,8 +274,7 @@ public abstract class AbstractBaseDispatcher extends EntryFilterSupport implemen
}
}
if (CollectionUtils.isNotEmpty(entriesToFiltered)) {
- subscription.acknowledgeMessage(entriesToFiltered, AckType.Individual,
- Collections.emptyMap());
+ individualAcknowledgeMessageIfNeeded(entriesToFiltered, Collections.emptyMap());
int filtered = entriesToFiltered.size();
Topic topic = subscription.getTopic();
@@ -301,9 +303,9 @@ public abstract class AbstractBaseDispatcher extends EntryFilterSupport implemen
return totalEntries;
}
- private void individualAcknowledgeMessageIfNeeded(Position position, Map<String, Long> properties) {
+ private void individualAcknowledgeMessageIfNeeded(List<Position> positions, Map<String, Long> properties) {
if (!(subscription instanceof PulsarCompactorSubscription)) {
- subscription.acknowledgeMessage(Collections.singletonList(position), AckType.Individual, properties);
+ subscription.acknowledgeMessage(positions, AckType.Individual, properties);
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/plugin/FilterEntryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/plugin/FilterEntryTest.java
index b868858646c..1c4f88bc027 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/plugin/FilterEntryTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/plugin/FilterEntryTest.java
@@ -22,6 +22,7 @@ import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructor
import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgsRecordingInvocations;
import static org.apache.pulsar.client.api.SubscriptionInitialPosition.Earliest;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
@@ -30,8 +31,9 @@ import static org.mockito.Mockito.when;
import static org.testng.Assert.assertTrue;
import static org.testng.AssertJUnit.assertEquals;
import static org.testng.AssertJUnit.assertNotNull;
-
+import io.netty.buffer.ByteBuf;
import java.lang.reflect.Field;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -39,7 +41,6 @@ import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
-
import lombok.Cleanup;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
@@ -58,11 +59,15 @@ import org.apache.pulsar.broker.testcontext.PulsarTestContext;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.RawMessage;
+import org.apache.pulsar.client.api.RawReader;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.common.nar.NarClassLoader;
+import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.stats.AnalyzeSubscriptionBacklogResult;
+import org.apache.pulsar.compaction.Compactor;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
@@ -150,6 +155,58 @@ public class FilterEntryTest extends BrokerTestBase {
consumer.close();
}
+ @Test
+ public void testEntryFilterWithCompactor() throws Exception {
+ conf.setAllowOverrideEntryFilters(true);
+ String topic = "persistent://prop/ns-abc/topic" + UUID.randomUUID();
+
+ List<String> messages = new ArrayList<>();
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .enableBatching(false).topic(topic).create();
+ producer.newMessage().key("K1").value("V1").send();
+ producer.newMessage().key("K2").value("V2").send();
+ producer.newMessage().key("K3").value("V3").send();
+ producer.newMessage().key("K4").value("V4").send();
+ messages.add("V2");
+ messages.add("V4");
+
+ PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic).get();
+
+ // set topic level entry filters
+ EntryFilter mockFilter = mock(EntryFilter.class);
+ doAnswer(invocationOnMock -> {
+ FilterContext filterContext = invocationOnMock.getArgument(1);
+ String partitionKey = filterContext.getMsgMetadata().getPartitionKey();
+ if (partitionKey.equals("K1") || partitionKey.equals("K3")) {
+ return EntryFilter.FilterResult.REJECT;
+ } else {
+ return EntryFilter.FilterResult.ACCEPT;
+ }
+ }).when(mockFilter).filterEntry(any(Entry.class), any(FilterContext.class));
+ setMockFilterToTopic(topicRef, List.of(mockFilter));
+
+ List<String> results = new ArrayList<>();
+ RawReader rawReader = RawReader.create(pulsarClient, topic, Compactor.COMPACTION_SUBSCRIPTION).get();
+ while (true) {
+ boolean hasMsg = rawReader.hasMessageAvailableAsync().get();
+ if (hasMsg) {
+ try (RawMessage m = rawReader.readNextAsync().get()) {
+ ByteBuf headersAndPayload = m.getHeadersAndPayload();
+ Commands.skipMessageMetadata(headersAndPayload);
+ byte[] bytes = new byte[headersAndPayload.readableBytes()];
+ headersAndPayload.readBytes(bytes);
+
+ results.add(new String(bytes));
+ }
+ } else {
+ break;
+ }
+ }
+ rawReader.closeAsync().get();
+
+ Assert.assertEquals(messages, results);
+ }
+
@SneakyThrows
private void setMockFilterToTopic(PersistentTopic topicRef, List<EntryFilter> mockFilter) {
FieldUtils.writeField(topicRef, "entryFilters", Pair.of(null, mockFilter), true);