From 568671778b12eded01751f2b6ecf375246ff3d2e Mon Sep 17 00:00:00 2001 From: rangao Date: Mon, 27 Feb 2023 09:18:52 +0800 Subject: [PATCH 1/7] reuse transaction buffer snapshot writer for topics which belong to the same namespace. --- .../SystemTopicTxnBufferSnapshotService.java | 116 +++++++++++++++--- ...SingleSnapshotAbortedTxnProcessorImpl.java | 21 ++-- ...napshotSegmentAbortedTxnProcessorImpl.java | 39 +++--- .../TopicTransactionBufferRecoverTest.java | 11 +- .../broker/transaction/TransactionTest.java | 79 ++++++++++++ .../transaction/TransactionTestBase.java | 20 +++ .../buffer/TransactionBufferCloseTest.java | 79 ++++++------ 7 files changed, 288 insertions(+), 77 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java index a1b78d89a13eb..4c1938a5feb6c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java @@ -18,66 +18,154 @@ */ package org.apache.pulsar.broker.service; +import java.util.HashMap; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; +import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory; import org.apache.pulsar.broker.systopic.SystemTopicClient; import org.apache.pulsar.broker.systopic.SystemTopicClientBase; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.common.events.EventType; +import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.naming.TopicName; -import org.apache.pulsar.common.util.FutureUtil; +@Slf4j public class SystemTopicTxnBufferSnapshotService { - protected final Map> clients; + protected final ConcurrentHashMap> clients; protected final NamespaceEventsSystemTopicFactory namespaceEventsSystemTopicFactory; protected final Class schemaType; protected final EventType systemTopicType; + private final HashMap> refCountedWriterMap; + + // The class ReferenceCountedWriter will maintain the reference count, + // when the reference count decrement to 0, it will be removed from writerFutureMap, the writer will be closed. + public static class ReferenceCountedWriter { + + private final AtomicLong referenceCount; + private final NamespaceName namespaceName; + private final CompletableFuture> future; + private final SystemTopicTxnBufferSnapshotService snapshotService; + + public ReferenceCountedWriter(NamespaceName namespaceName, + CompletableFuture> future, + SystemTopicTxnBufferSnapshotService snapshotService) { + this.referenceCount = new AtomicLong(1); + this.namespaceName = namespaceName; + this.snapshotService = snapshotService; + this.future = future; + this.future.exceptionally(t -> { + log.error("[{}] Failed to create transaction buffer snapshot writer.", namespaceName, t); + snapshotService.refCountedWriterMap.remove(namespaceName, this); + return null; + }); + } + + public CompletableFuture> getFuture() { + return future; + } + + private void retain() { + operationValidate(true); + this.referenceCount.incrementAndGet(); + } + + private long release() { + operationValidate(false); + return this.referenceCount.decrementAndGet(); + } + + private void operationValidate(boolean isRetain) { + if (this.referenceCount.get() == 0) { + throw new RuntimeException( + "[" + namespaceName + "] The reference counted transaction buffer snapshot writer couldn't " + + "be " + (isRetain ? "retained" : "released") + ", refCnt is 0."); + } + } + + public void close() { + if (release() == 0) { + snapshotService.refCountedWriterMap.remove(namespaceName, this); + future.thenAccept(writer -> { + final String topicName = writer.getSystemTopicClient().getTopicName().toString(); + writer.closeAsync().exceptionally(t -> { + if (t != null) { + log.error("[{}] Failed to close writer.", topicName, t); + } else { + if (log.isDebugEnabled()) { + log.debug("[{}] Success to close writer.", topicName); + } + } + return null; + }); + }); + } + } + + } + public SystemTopicTxnBufferSnapshotService(PulsarClient client, EventType systemTopicType, Class schemaType) { this.namespaceEventsSystemTopicFactory = new NamespaceEventsSystemTopicFactory(client); this.systemTopicType = systemTopicType; this.schemaType = schemaType; this.clients = new ConcurrentHashMap<>(); + this.refCountedWriterMap = new HashMap<>(); } public CompletableFuture> createWriter(TopicName topicName) { - return getTransactionBufferSystemTopicClient(topicName).thenCompose(SystemTopicClient::newWriterAsync); + return getTransactionBufferSystemTopicClient(topicName).newWriterAsync(); } public CompletableFuture> createReader(TopicName topicName) { - return getTransactionBufferSystemTopicClient(topicName).thenCompose(SystemTopicClient::newReaderAsync); + return getTransactionBufferSystemTopicClient(topicName).newReaderAsync(); } public void removeClient(TopicName topicName, SystemTopicClientBase transactionBufferSystemTopicClient) { if (transactionBufferSystemTopicClient.getReaders().size() == 0 && transactionBufferSystemTopicClient.getWriters().size() == 0) { - clients.remove(topicName); + clients.remove(topicName.getNamespaceObject()); } } - protected CompletableFuture> getTransactionBufferSystemTopicClient(TopicName topicName) { + public synchronized ReferenceCountedWriter getReferenceWriter(TopicName topicName) { + return refCountedWriterMap.compute(topicName.getNamespaceObject(), (k, v) -> { + if (v == null) { + return new ReferenceCountedWriter<>(topicName.getNamespaceObject(), + getTransactionBufferSystemTopicClient(topicName).newWriterAsync(), this); + } else { + v.retain(); + } + return v; + }); + } + + private SystemTopicClient getTransactionBufferSystemTopicClient(TopicName topicName) { + if (topicName == null) { + throw new RuntimeException(new PulsarClientException + .InvalidTopicNameException("Can't get the tb system topic client due to the topic name is null.")); + } TopicName systemTopicName = NamespaceEventsSystemTopicFactory .getSystemTopicName(topicName.getNamespaceObject(), systemTopicType); if (systemTopicName == null) { - return FutureUtil.failedFuture( - new PulsarClientException - .InvalidTopicNameException("Can't create SystemTopicBaseTxnBufferSnapshotIndexService, " - + "because the topicName is null!")); + throw new RuntimeException(new PulsarClientException.InvalidTopicNameException( + "Can't get the tb system topic client for topic " + topicName + + " with type " + systemTopicType + ".")); } - return CompletableFuture.completedFuture(clients.computeIfAbsent(systemTopicName, + + return clients.computeIfAbsent(topicName.getNamespaceObject(), (v) -> namespaceEventsSystemTopicFactory - .createTransactionBufferSystemTopicClient(systemTopicName, - this, schemaType))); + .createTransactionBufferSystemTopicClient(systemTopicName, this, schemaType)); } public void close() throws Exception { - for (Map.Entry> entry : clients.entrySet()) { + for (Map.Entry> entry : clients.entrySet()) { entry.getValue().close(); } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java index 87161e97512b9..3bcaeb4dddd4a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java @@ -28,6 +28,7 @@ import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.commons.collections4.map.LinkedMap; import org.apache.pulsar.broker.service.BrokerServiceException; +import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService.ReferenceCountedWriter; import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.broker.systopic.SystemTopicClient; import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor; @@ -42,7 +43,7 @@ @Slf4j public class SingleSnapshotAbortedTxnProcessorImpl implements AbortedTxnProcessor { private final PersistentTopic topic; - private final CompletableFuture> takeSnapshotWriter; + private final ReferenceCountedWriter takeSnapshotWriter; /** * Aborts, map for jude message is aborted, linked for remove abort txn in memory when this * position have been deleted. @@ -51,12 +52,14 @@ public class SingleSnapshotAbortedTxnProcessorImpl implements AbortedTxnProcesso private volatile long lastSnapshotTimestamps; + private volatile boolean isClosed = false; + public SingleSnapshotAbortedTxnProcessorImpl(PersistentTopic topic) { this.topic = topic; this.takeSnapshotWriter = this.topic.getBrokerService().getPulsar() .getTransactionBufferSnapshotServiceFactory() - .getTxnBufferSnapshotService().createWriter(TopicName.get(topic.getName())); - this.takeSnapshotWriter.exceptionally((ex) -> { + .getTxnBufferSnapshotService().getReferenceWriter(TopicName.get(topic.getName())); + this.takeSnapshotWriter.getFuture().exceptionally((ex) -> { log.error("{} Failed to create snapshot writer", topic.getName()); topic.close(); return null; @@ -132,7 +135,7 @@ public CompletableFuture recoverFromSnapshot() { @Override public CompletableFuture clearAbortedTxnSnapshot() { - return this.takeSnapshotWriter.thenCompose(writer -> { + return this.takeSnapshotWriter.getFuture().thenCompose(writer -> { TransactionBufferSnapshot snapshot = new TransactionBufferSnapshot(); snapshot.setTopicName(topic.getName()); return writer.deleteAsync(snapshot.getTopicName(), snapshot); @@ -141,7 +144,7 @@ public CompletableFuture clearAbortedTxnSnapshot() { @Override public CompletableFuture takeAbortedTxnsSnapshot(PositionImpl maxReadPosition) { - return takeSnapshotWriter.thenCompose(writer -> { + return takeSnapshotWriter.getFuture().thenCompose(writer -> { TransactionBufferSnapshot snapshot = new TransactionBufferSnapshot(); snapshot.setTopicName(topic.getName()); snapshot.setMaxReadPositionLedgerId(maxReadPosition.getLedgerId()); @@ -175,8 +178,12 @@ public long getLastSnapshotTimestamps() { } @Override - public CompletableFuture closeAsync() { - return takeSnapshotWriter.thenCompose(SystemTopicClient.Writer::closeAsync); + public synchronized CompletableFuture closeAsync() { + if (!isClosed) { + isClosed = true; + takeSnapshotWriter.close(); + } + return CompletableFuture.completedFuture(null); } private void closeReader(SystemTopicClient.Reader reader) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java index 7a9e0e1abedd9..55a44ae9098a0 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java @@ -43,6 +43,7 @@ import org.apache.commons.lang3.tuple.MutablePair; import org.apache.commons.lang3.tuple.Pair; import org.apache.pulsar.broker.service.BrokerServiceException; +import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService.ReferenceCountedWriter; import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.broker.systopic.SystemTopicClient; import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor; @@ -442,11 +443,13 @@ public class PersistentWorker { private final PersistentTopic topic; //Persistent snapshot segment and index at the single thread. - private final CompletableFuture> + private final ReferenceCountedWriter snapshotSegmentsWriterFuture; - private final CompletableFuture> + private final ReferenceCountedWriter snapshotIndexWriterFuture; + private volatile boolean closed = false; + private enum OperationState { None, Operating, @@ -470,18 +473,18 @@ public enum OperationType { public PersistentWorker(PersistentTopic topic) { this.topic = topic; - this.snapshotSegmentsWriterFuture = this.topic.getBrokerService().getPulsar() + this.snapshotSegmentsWriterFuture = this.topic.getBrokerService().getPulsar() .getTransactionBufferSnapshotServiceFactory() - .getTxnBufferSnapshotSegmentService().createWriter(TopicName.get(topic.getName())); - this.snapshotSegmentsWriterFuture.exceptionally(ex -> { + .getTxnBufferSnapshotSegmentService().getReferenceWriter(TopicName.get(topic.getName())); + this.snapshotSegmentsWriterFuture.getFuture().exceptionally(ex -> { log.error("{} Failed to create snapshot index writer", topic.getName()); topic.close(); return null; }); this.snapshotIndexWriterFuture = this.topic.getBrokerService().getPulsar() .getTransactionBufferSnapshotServiceFactory() - .getTxnBufferSnapshotIndexService().createWriter(TopicName.get(topic.getName())); - this.snapshotIndexWriterFuture.exceptionally((ex) -> { + .getTxnBufferSnapshotIndexService().getReferenceWriter(TopicName.get(topic.getName())); + this.snapshotIndexWriterFuture.getFuture().exceptionally((ex) -> { log.error("{} Failed to create snapshot writer", topic.getName()); topic.close(); return null; @@ -631,7 +634,7 @@ private CompletableFuture writeSnapshotSegmentAsync(LinkedList segm transactionBufferSnapshotSegment.setPersistentPositionLedgerId( abortedMarkerPersistentPosition.getLedgerId()); - return snapshotSegmentsWriterFuture.thenCompose(segmentWriter -> { + return snapshotSegmentsWriterFuture.getFuture().thenCompose(segmentWriter -> { transactionBufferSnapshotSegment.setSequenceId(this.sequenceID.get()); return segmentWriter.writeAsync(buildKey(this.sequenceID.get()), transactionBufferSnapshotSegment); }).thenCompose((messageId) -> { @@ -668,7 +671,7 @@ private CompletableFuture deleteSnapshotSegment(List positio List> results = new ArrayList<>(); for (PositionImpl positionNeedToDelete : positionNeedToDeletes) { long sequenceIdNeedToDelete = indexes.get(positionNeedToDelete).getSequenceID(); - CompletableFuture res = snapshotSegmentsWriterFuture + CompletableFuture res = snapshotSegmentsWriterFuture.getFuture() .thenCompose(writer -> writer.deleteAsync(buildKey(sequenceIdNeedToDelete), null)) .thenCompose(messageId -> { if (log.isDebugEnabled()) { @@ -695,7 +698,7 @@ private CompletableFuture deleteSnapshotSegment(List positio private CompletableFuture updateSnapshotIndex(TransactionBufferSnapshotIndexesMetadata snapshotSegment) { TransactionBufferSnapshotIndexes snapshotIndexes = new TransactionBufferSnapshotIndexes(); - CompletableFuture res = snapshotIndexWriterFuture + CompletableFuture res = snapshotIndexWriterFuture.getFuture() .thenCompose((indexesWriter) -> { snapshotIndexes.setIndexList(indexes.values().stream().toList()); snapshotIndexes.setSnapshot(snapshotSegment); @@ -712,7 +715,7 @@ private CompletableFuture updateSnapshotIndex(TransactionBufferSnapshotInd private CompletableFuture clearSnapshotSegmentAndIndexes() { CompletableFuture res = persistentWorker.clearAllSnapshotSegments() - .thenCompose((ignore) -> snapshotIndexWriterFuture + .thenCompose((ignore) -> snapshotIndexWriterFuture.getFuture() .thenCompose(indexesWriter -> indexesWriter.writeAsync(topic.getName(), null))) .thenRun(() -> log.debug("Successes to clear the snapshot segment and indexes for the topic [{}]", @@ -747,7 +750,7 @@ private CompletableFuture clearAllSnapshotSegments() { Message message = reader.readNextAsync() .get(getSystemClientOperationTimeoutMs(), TimeUnit.MILLISECONDS); if (topic.getName().equals(message.getValue().getTopicName())) { - snapshotSegmentsWriterFuture.get().write(message.getKey(), null); + snapshotSegmentsWriterFuture.getFuture().get().write(message.getKey(), null); } } return CompletableFuture.completedFuture(null); @@ -760,11 +763,13 @@ private CompletableFuture clearAllSnapshotSegments() { }); } - - CompletableFuture closeAsync() { - return CompletableFuture.allOf( - this.snapshotIndexWriterFuture.thenCompose(SystemTopicClient.Writer::closeAsync), - this.snapshotSegmentsWriterFuture.thenCompose(SystemTopicClient.Writer::closeAsync)); + synchronized CompletableFuture closeAsync() { + if (!closed) { + closed = true; + this.snapshotIndexWriterFuture.close(); + this.snapshotSegmentsWriterFuture.close(); + } + return CompletableFuture.completedFuture(null); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java index d4ddb26e014ca..45ad6e4c58254 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java @@ -57,6 +57,7 @@ import org.apache.pulsar.broker.service.AbstractTopic; import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService; +import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService.ReferenceCountedWriter; import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.broker.service.TransactionBufferSnapshotServiceFactory; import org.apache.pulsar.broker.service.persistent.PersistentTopic; @@ -602,11 +603,12 @@ public void testTransactionBufferRecoverThrowException() throws Exception { mock(SystemTopicTxnBufferSnapshotService.class); SystemTopicClient.Reader reader = mock(SystemTopicClient.Reader.class); SystemTopicClient.Writer writer = mock(SystemTopicClient.Writer.class); + ReferenceCountedWriter refCounterWriter = mock(ReferenceCountedWriter.class); + doReturn(CompletableFuture.completedFuture(writer)).when(refCounterWriter).getFuture(); doReturn(CompletableFuture.completedFuture(reader)) .when(systemTopicTxnBufferSnapshotService).createReader(any()); - doReturn(CompletableFuture.completedFuture(writer)) - .when(systemTopicTxnBufferSnapshotService).createWriter(any()); + doReturn(refCounterWriter).when(systemTopicTxnBufferSnapshotService).getReferenceWriter(any()); TransactionBufferSnapshotServiceFactory transactionBufferSnapshotServiceFactory = mock(TransactionBufferSnapshotServiceFactory.class); doReturn(systemTopicTxnBufferSnapshotService) @@ -645,8 +647,9 @@ public void testTransactionBufferRecoverThrowException() throws Exception { originalTopic = (PersistentTopic) getPulsarServiceList().get(0) .getBrokerService().getTopic(TopicName.get(topic).toString(), false).get().get(); // mock create writer fail - doReturn(FutureUtil.failedFuture(new PulsarClientException("test"))) - .when(systemTopicTxnBufferSnapshotService).createWriter(any()); + ReferenceCountedWriter failedCountedWriter = mock(ReferenceCountedWriter.class); + doReturn(FutureUtil.failedFuture(new PulsarClientException("test"))).when(failedCountedWriter).getFuture(); + doReturn(failedCountedWriter).when(systemTopicTxnBufferSnapshotService).getReferenceWriter(any()); checkCloseTopic(pulsarClient, transactionBufferSnapshotServiceFactoryOriginal, transactionBufferSnapshotServiceFactory, originalTopic, field, producer); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java index 20aeac0ed648f..1e8a1e17ad877 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java @@ -86,6 +86,7 @@ import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.broker.service.BrokerServiceException; import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService; +import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService.ReferenceCountedWriter; import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.broker.service.TransactionBufferSnapshotServiceFactory; import org.apache.pulsar.broker.service.persistent.PersistentSubscription; @@ -1676,4 +1677,82 @@ public void testDeleteNamespace() throws Exception { admin.namespaces().deleteNamespace(namespace, true); } + + @Test(timeOut = 10_000) + public void testTBSnapshotWriter() throws Exception { + String namespace = TENANT + "/ns-" + RandomStringUtils.randomAlphabetic(5); + admin.namespaces().createNamespace(namespace, 16); + String topic = namespace + "/test-create-snapshot-writer-failed"; + int partitionCount = 20; + admin.topics().createPartitionedTopic(topic, partitionCount); + + Class clazz = SystemTopicTxnBufferSnapshotService.class; + Field field = clazz.getDeclaredField("writerMap"); + field.setAccessible(true); + // inject a failed writer future + CompletableFuture> writerFuture = new CompletableFuture<>(); + for (PulsarService pulsarService : pulsarServiceList) { + SystemTopicTxnBufferSnapshotService bufferSnapshotService = + pulsarService.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotService(); + HashMap writerMap1 = + ((HashMap) field.get(bufferSnapshotService)); + ReferenceCountedWriter failedCountedWriter = + new ReferenceCountedWriter(NamespaceName.get(namespace), writerFuture, bufferSnapshotService); + writerMap1.put(NamespaceName.get(namespace), failedCountedWriter); + + SystemTopicTxnBufferSnapshotService segmentSnapshotService = + pulsarService.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotSegmentService(); + HashMap writerMap2 = + ((HashMap) field.get(segmentSnapshotService)); + ReferenceCountedWriter failedCountedWriter2 = + new ReferenceCountedWriter(NamespaceName.get(namespace), writerFuture, segmentSnapshotService); + writerMap2.put(NamespaceName.get(namespace), failedCountedWriter2); + + SystemTopicTxnBufferSnapshotService indexSnapshotService = + pulsarService.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotIndexService(); + HashMap writerMap3 = + ((HashMap) field.get(indexSnapshotService)); + ReferenceCountedWriter failedCountedWriter3 = + new ReferenceCountedWriter(NamespaceName.get(namespace), writerFuture, indexSnapshotService); + writerMap3.put(NamespaceName.get(namespace), failedCountedWriter3); + } + + CompletableFuture> producerFuture = pulsarClient.newProducer() + .topic(topic) + .sendTimeout(0, TimeUnit.SECONDS) + .createAsync(); + getTopic("persistent://" + topic + "-partition-0"); + Thread.sleep(3000); + // the producer shouldn't be created, because the transaction buffer snapshot writer future didn't finish. + assertFalse(producerFuture.isDone()); + + // The topic will be closed, because the transaction buffer snapshot writer future is failed, + // the failed writer future will be removed, the producer will be reconnected and work well. + writerFuture.completeExceptionally(new PulsarClientException.TopicTerminatedException("failed writer")); + Producer producer = producerFuture.get(); + + for (int i = 0; i < partitionCount * 2; i++) { + Transaction txn = pulsarClient.newTransaction() + .withTransactionTimeout(1, TimeUnit.MINUTES).build().get(); + producer.newMessage(txn).value("test".getBytes()).sendAsync(); + txn.commit().get(); + } + checkSnapshotPublisherCount(namespace, 1); + producer.close(); + admin.topics().unload(topic); + checkSnapshotPublisherCount(namespace, 0); + } + + private void getTopic(String topicName) { + Awaitility.await().atMost(5, TimeUnit.SECONDS).pollInterval(100, TimeUnit.MILLISECONDS) + .until(() -> { + for (PulsarService pulsarService : pulsarServiceList) { + if (pulsarService.getBrokerService().getTopicReference(topicName).isPresent()) { + return true; + } + } + return false; + }); + } + } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java index fd49354342fa0..f45eda8d21fbe 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java @@ -41,12 +41,17 @@ import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.naming.SystemTopicNames; +import org.apache.pulsar.common.naming.TopicDomain; +import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.policies.data.ClusterData; +import org.apache.pulsar.common.policies.data.PublisherStats; import org.apache.pulsar.common.policies.data.TenantInfoImpl; import org.apache.pulsar.common.policies.data.TopicType; import org.apache.pulsar.metadata.api.MetadataStoreException; import org.apache.pulsar.tests.TestRetrySupport; +import org.awaitility.Awaitility; +import org.testng.Assert; @Slf4j public abstract class TransactionTestBase extends TestRetrySupport { @@ -223,4 +228,19 @@ protected void deleteNamespaceWithRetry(String ns, boolean force, PulsarAdmin ad throws Exception { MockedPulsarServiceBaseTest.deleteNamespaceWithRetry(ns, force, admin, pulsarServiceList); } + + public void checkSnapshotPublisherCount(String namespace, int expectCount) { + TopicName snTopicName = TopicName.get(TopicDomain.persistent.value(), NamespaceName.get(namespace), + SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT); + Awaitility.await() + .atMost(5, TimeUnit.SECONDS) + .pollInterval(100, TimeUnit.MILLISECONDS) + .untilAsserted(() -> { + List publisherStatsList = + (List) admin.topics() + .getStats(snTopicName.getPartitionedTopicName()).getPublishers(); + Assert.assertEquals(publisherStatsList.size(), expectCount); + }); + } + } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferCloseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferCloseTest.java index cd3a14da5967d..e92cf29521e34 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferCloseTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferCloseTest.java @@ -19,6 +19,9 @@ package org.apache.pulsar.broker.transaction.buffer; import com.google.common.collect.Sets; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.RandomStringUtils; import org.apache.pulsar.broker.transaction.TransactionTestBase; @@ -26,20 +29,13 @@ import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClient; import org.apache.pulsar.client.impl.PulsarClientImpl; -import org.apache.pulsar.common.naming.NamespaceName; -import org.apache.pulsar.common.naming.SystemTopicNames; -import org.apache.pulsar.common.naming.TopicDomain; import org.apache.pulsar.common.naming.TopicName; -import org.apache.pulsar.common.policies.data.PublisherStats; import org.apache.pulsar.common.policies.data.TenantInfoImpl; import org.awaitility.Awaitility; -import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; -import java.util.List; -import java.util.concurrent.TimeUnit; /** * Transaction buffer close test. @@ -71,49 +67,62 @@ public Object[][] isPartition() { @Test(timeOut = 10_000, dataProvider = "isPartition") public void deleteTopicCloseTransactionBufferTest(boolean isPartition) throws Exception { - int expectedCount = isPartition ? 30 : 1; - TopicName topicName = createAndLoadTopic(isPartition, expectedCount); - checkSnapshotPublisherCount(topicName.getNamespace(), expectedCount); + int partitionCount = isPartition ? 30 : 1; + List topicNames = createAndLoadTopics(isPartition, partitionCount); + String namespaceName = topicNames.get(0).getNamespace(); + checkSnapshotPublisherCount(namespaceName, 1); + + for (int i = 0; i < topicNames.size(); i++) { + deleteTopic(isPartition, topicNames.get(i)); + // When delete all topics of the namespace, the publisher count should be 0. + int expectCount = i == topicNames.size() - 1 ? 0 : 1; + checkSnapshotPublisherCount(namespaceName, expectCount); + } + } + + private void deleteTopic(boolean isPartition, TopicName topicName) throws PulsarAdminException { if (isPartition) { admin.topics().deletePartitionedTopic(topicName.getPartitionedTopicName(), true); } else { admin.topics().delete(topicName.getPartitionedTopicName(), true); } - checkSnapshotPublisherCount(topicName.getNamespace(), 0); } @Test(timeOut = 10_000, dataProvider = "isPartition") public void unloadTopicCloseTransactionBufferTest(boolean isPartition) throws Exception { - int expectedCount = isPartition ? 30 : 1; - TopicName topicName = createAndLoadTopic(isPartition, expectedCount); - checkSnapshotPublisherCount(topicName.getNamespace(), expectedCount); - admin.topics().unload(topicName.getPartitionedTopicName()); - checkSnapshotPublisherCount(topicName.getNamespace(), 0); + int partitionCount = isPartition ? 30 : 1; + List topicNames = createAndLoadTopics(isPartition, partitionCount); + String namespaceName = topicNames.get(0).getNamespace(); + checkSnapshotPublisherCount(namespaceName, 1); + + for (int i = 0; i < topicNames.size(); i++) { + admin.topics().unload(topicNames.get(i).getPartitionedTopicName()); + // When unload all topics of the namespace, the publisher count should be 0. + int expectCount = i == topicNames.size() - 1 ? 0 : 1; + checkSnapshotPublisherCount(namespaceName, expectCount); + } } - private TopicName createAndLoadTopic(boolean isPartition, int partitionCount) + private List createAndLoadTopics(boolean isPartition, int partitionCount) throws PulsarAdminException, PulsarClientException { String namespace = TENANT + "/ns-" + RandomStringUtils.randomAlphabetic(5); admin.namespaces().createNamespace(namespace, 3); - String topic = namespace + "/tb-close-test-"; - if (isPartition) { - admin.topics().createPartitionedTopic(topic, partitionCount); - } - pulsarClient.newProducer() - .topic(topic) - .sendTimeout(0, TimeUnit.SECONDS) - .create() - .close(); - return TopicName.get(topic); - } + String topic = namespace + "/tb-close-test"; + List topics = new ArrayList<>(); - private void checkSnapshotPublisherCount(String namespace, int expectCount) throws PulsarAdminException { - TopicName snTopicName = TopicName.get(TopicDomain.persistent.value(), NamespaceName.get(namespace), - SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT); - List publisherStatsList = - (List) admin.topics() - .getStats(snTopicName.getPartitionedTopicName()).getPublishers(); - Assert.assertEquals(publisherStatsList.size(), expectCount); + for (int i = 0; i < 2; i++) { + String t = topic + "-" + i; + if (isPartition) { + admin.topics().createPartitionedTopic(t, partitionCount); + } + pulsarClient.newProducer() + .topic(t) + .sendTimeout(0, TimeUnit.SECONDS) + .create() + .close(); + topics.add(TopicName.get(t)); + } + return topics; } } From fee495afb86a1d23343644dce4f8815d2def40b6 Mon Sep 17 00:00:00 2001 From: rangao Date: Tue, 28 Feb 2023 15:57:46 +0800 Subject: [PATCH 2/7] address comment --- .../service/SystemTopicTxnBufferSnapshotService.java | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java index 4c1938a5feb6c..041195a6c7ce9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java @@ -18,7 +18,6 @@ */ package org.apache.pulsar.broker.service; -import java.util.HashMap; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -42,7 +41,7 @@ public class SystemTopicTxnBufferSnapshotService { protected final Class schemaType; protected final EventType systemTopicType; - private final HashMap> refCountedWriterMap; + private final ConcurrentHashMap> refCountedWriterMap; // The class ReferenceCountedWriter will maintain the reference count, // when the reference count decrement to 0, it will be removed from writerFutureMap, the writer will be closed. @@ -116,7 +115,7 @@ public SystemTopicTxnBufferSnapshotService(PulsarClient client, EventType system this.systemTopicType = systemTopicType; this.schemaType = schemaType; this.clients = new ConcurrentHashMap<>(); - this.refCountedWriterMap = new HashMap<>(); + this.refCountedWriterMap = new ConcurrentHashMap<>(); } public CompletableFuture> createWriter(TopicName topicName) { @@ -134,7 +133,7 @@ public void removeClient(TopicName topicName, SystemTopicClientBase transacti } } - public synchronized ReferenceCountedWriter getReferenceWriter(TopicName topicName) { + public ReferenceCountedWriter getReferenceWriter(TopicName topicName) { return refCountedWriterMap.compute(topicName.getNamespaceObject(), (k, v) -> { if (v == null) { return new ReferenceCountedWriter<>(topicName.getNamespaceObject(), From e28e732c49c6b4d0471287eaba2ba6b6d737e0b8 Mon Sep 17 00:00:00 2001 From: rangao Date: Thu, 2 Mar 2023 18:10:00 +0800 Subject: [PATCH 3/7] address comment --- .../SystemTopicTxnBufferSnapshotService.java | 39 +++++-------------- ...SingleSnapshotAbortedTxnProcessorImpl.java | 2 +- ...napshotSegmentAbortedTxnProcessorImpl.java | 28 +++++++------ .../TopicTransactionBufferRecoverTest.java | 6 ++- .../broker/transaction/TransactionTest.java | 6 ++- 5 files changed, 32 insertions(+), 49 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java index 041195a6c7ce9..abf579db309d6 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java @@ -70,41 +70,27 @@ public CompletableFuture> getFuture() { return future; } - private void retain() { - operationValidate(true); - this.referenceCount.incrementAndGet(); + private synchronized boolean retain() { + return this.referenceCount.incrementAndGet() > 0; } - private long release() { - operationValidate(false); - return this.referenceCount.decrementAndGet(); - } - - private void operationValidate(boolean isRetain) { - if (this.referenceCount.get() == 0) { - throw new RuntimeException( - "[" + namespaceName + "] The reference counted transaction buffer snapshot writer couldn't " - + "be " + (isRetain ? "retained" : "released") + ", refCnt is 0."); - } - } - - public void close() { - if (release() == 0) { + public synchronized void release() { + if (this.referenceCount.decrementAndGet() <= 0) { snapshotService.refCountedWriterMap.remove(namespaceName, this); future.thenAccept(writer -> { final String topicName = writer.getSystemTopicClient().getTopicName().toString(); writer.closeAsync().exceptionally(t -> { if (t != null) { - log.error("[{}] Failed to close writer.", topicName, t); + log.error("[{}] Failed to close TB snapshot writer.", topicName, t); } else { if (log.isDebugEnabled()) { - log.debug("[{}] Success to close writer.", topicName); + log.debug("[{}] Success to close TB snapshot writer.", topicName); } } return null; }); }); - } + }; } } @@ -118,10 +104,6 @@ public SystemTopicTxnBufferSnapshotService(PulsarClient client, EventType system this.refCountedWriterMap = new ConcurrentHashMap<>(); } - public CompletableFuture> createWriter(TopicName topicName) { - return getTransactionBufferSystemTopicClient(topicName).newWriterAsync(); - } - public CompletableFuture> createReader(TopicName topicName) { return getTransactionBufferSystemTopicClient(topicName).newReaderAsync(); } @@ -135,13 +117,12 @@ public void removeClient(TopicName topicName, SystemTopicClientBase transacti public ReferenceCountedWriter getReferenceWriter(TopicName topicName) { return refCountedWriterMap.compute(topicName.getNamespaceObject(), (k, v) -> { - if (v == null) { + if (v != null && v.retain()) { + return v; + } else { return new ReferenceCountedWriter<>(topicName.getNamespaceObject(), getTransactionBufferSystemTopicClient(topicName).newWriterAsync(), this); - } else { - v.retain(); } - return v; }); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java index 3bcaeb4dddd4a..bb08a7f60309a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java @@ -181,7 +181,7 @@ public long getLastSnapshotTimestamps() { public synchronized CompletableFuture closeAsync() { if (!isClosed) { isClosed = true; - takeSnapshotWriter.close(); + takeSnapshotWriter.release(); } return CompletableFuture.completedFuture(null); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java index 55a44ae9098a0..64c9e2b393cae 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java @@ -443,10 +443,8 @@ public class PersistentWorker { private final PersistentTopic topic; //Persistent snapshot segment and index at the single thread. - private final ReferenceCountedWriter - snapshotSegmentsWriterFuture; - private final ReferenceCountedWriter - snapshotIndexWriterFuture; + private final ReferenceCountedWriter snapshotSegmentsWriter; + private final ReferenceCountedWriter snapshotIndexWriter; private volatile boolean closed = false; @@ -473,18 +471,18 @@ public enum OperationType { public PersistentWorker(PersistentTopic topic) { this.topic = topic; - this.snapshotSegmentsWriterFuture = this.topic.getBrokerService().getPulsar() + this.snapshotSegmentsWriter = this.topic.getBrokerService().getPulsar() .getTransactionBufferSnapshotServiceFactory() .getTxnBufferSnapshotSegmentService().getReferenceWriter(TopicName.get(topic.getName())); - this.snapshotSegmentsWriterFuture.getFuture().exceptionally(ex -> { + this.snapshotSegmentsWriter.getFuture().exceptionally(ex -> { log.error("{} Failed to create snapshot index writer", topic.getName()); topic.close(); return null; }); - this.snapshotIndexWriterFuture = this.topic.getBrokerService().getPulsar() + this.snapshotIndexWriter = this.topic.getBrokerService().getPulsar() .getTransactionBufferSnapshotServiceFactory() .getTxnBufferSnapshotIndexService().getReferenceWriter(TopicName.get(topic.getName())); - this.snapshotIndexWriterFuture.getFuture().exceptionally((ex) -> { + this.snapshotIndexWriter.getFuture().exceptionally((ex) -> { log.error("{} Failed to create snapshot writer", topic.getName()); topic.close(); return null; @@ -634,7 +632,7 @@ private CompletableFuture writeSnapshotSegmentAsync(LinkedList segm transactionBufferSnapshotSegment.setPersistentPositionLedgerId( abortedMarkerPersistentPosition.getLedgerId()); - return snapshotSegmentsWriterFuture.getFuture().thenCompose(segmentWriter -> { + return snapshotSegmentsWriter.getFuture().thenCompose(segmentWriter -> { transactionBufferSnapshotSegment.setSequenceId(this.sequenceID.get()); return segmentWriter.writeAsync(buildKey(this.sequenceID.get()), transactionBufferSnapshotSegment); }).thenCompose((messageId) -> { @@ -671,7 +669,7 @@ private CompletableFuture deleteSnapshotSegment(List positio List> results = new ArrayList<>(); for (PositionImpl positionNeedToDelete : positionNeedToDeletes) { long sequenceIdNeedToDelete = indexes.get(positionNeedToDelete).getSequenceID(); - CompletableFuture res = snapshotSegmentsWriterFuture.getFuture() + CompletableFuture res = snapshotSegmentsWriter.getFuture() .thenCompose(writer -> writer.deleteAsync(buildKey(sequenceIdNeedToDelete), null)) .thenCompose(messageId -> { if (log.isDebugEnabled()) { @@ -698,7 +696,7 @@ private CompletableFuture deleteSnapshotSegment(List positio private CompletableFuture updateSnapshotIndex(TransactionBufferSnapshotIndexesMetadata snapshotSegment) { TransactionBufferSnapshotIndexes snapshotIndexes = new TransactionBufferSnapshotIndexes(); - CompletableFuture res = snapshotIndexWriterFuture.getFuture() + CompletableFuture res = snapshotIndexWriter.getFuture() .thenCompose((indexesWriter) -> { snapshotIndexes.setIndexList(indexes.values().stream().toList()); snapshotIndexes.setSnapshot(snapshotSegment); @@ -715,7 +713,7 @@ private CompletableFuture updateSnapshotIndex(TransactionBufferSnapshotInd private CompletableFuture clearSnapshotSegmentAndIndexes() { CompletableFuture res = persistentWorker.clearAllSnapshotSegments() - .thenCompose((ignore) -> snapshotIndexWriterFuture.getFuture() + .thenCompose((ignore) -> snapshotIndexWriter.getFuture() .thenCompose(indexesWriter -> indexesWriter.writeAsync(topic.getName(), null))) .thenRun(() -> log.debug("Successes to clear the snapshot segment and indexes for the topic [{}]", @@ -750,7 +748,7 @@ private CompletableFuture clearAllSnapshotSegments() { Message message = reader.readNextAsync() .get(getSystemClientOperationTimeoutMs(), TimeUnit.MILLISECONDS); if (topic.getName().equals(message.getValue().getTopicName())) { - snapshotSegmentsWriterFuture.getFuture().get().write(message.getKey(), null); + snapshotSegmentsWriter.getFuture().get().write(message.getKey(), null); } } return CompletableFuture.completedFuture(null); @@ -766,8 +764,8 @@ private CompletableFuture clearAllSnapshotSegments() { synchronized CompletableFuture closeAsync() { if (!closed) { closed = true; - this.snapshotIndexWriterFuture.close(); - this.snapshotSegmentsWriterFuture.close(); + snapshotSegmentsWriter.release(); + snapshotIndexWriter.release(); } return CompletableFuture.completedFuture(null); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java index 45ad6e4c58254..a2de13c0a3de5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java @@ -706,7 +706,8 @@ public void testTransactionBufferIndexSystemTopic() throws Exception { new TransactionBufferSnapshotServiceFactory(pulsarClient).getTxnBufferSnapshotIndexService(); SystemTopicClient.Writer indexesWriter = - transactionBufferSnapshotIndexService.createWriter(TopicName.get(SNAPSHOT_INDEX)).get(); + transactionBufferSnapshotIndexService.getReferenceWriter( + TopicName.get(SNAPSHOT_INDEX)).getFuture().get(); SystemTopicClient.Reader indexesReader = transactionBufferSnapshotIndexService.createReader(TopicName.get(SNAPSHOT_INDEX)).get(); @@ -767,7 +768,8 @@ public void testTransactionBufferSegmentSystemTopic() throws Exception { new TransactionBufferSnapshotServiceFactory(pulsarClient).getTxnBufferSnapshotSegmentService(); SystemTopicClient.Writer - segmentWriter = transactionBufferSnapshotSegmentService.createWriter(snapshotSegmentTopicName).get(); + segmentWriter = transactionBufferSnapshotSegmentService + .getReferenceWriter(snapshotSegmentTopicName).getFuture().get(); // write two snapshot to snapshot segment topic TransactionBufferSnapshotSegment snapshot = diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java index 1e8a1e17ad877..8e7288c899db6 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java @@ -1533,8 +1533,10 @@ public void testTBRecoverChangeStateError() throws InterruptedException, Timeout = mock(SystemTopicTxnBufferSnapshotService.class); SystemTopicClient.Writer writer = mock(SystemTopicClient.Writer.class); when(writer.closeAsync()).thenReturn(CompletableFuture.completedFuture(null)); - when(systemTopicTxnBufferSnapshotService.createWriter(any())) - .thenReturn(CompletableFuture.completedFuture(writer)); + ReferenceCountedWriter refCounterWriter = mock(ReferenceCountedWriter.class); + doReturn(CompletableFuture.completedFuture(writer)).when(refCounterWriter).getFuture(); + when(systemTopicTxnBufferSnapshotService.getReferenceWriter(any())) + .thenReturn(refCounterWriter); TransactionBufferSnapshotServiceFactory transactionBufferSnapshotServiceFactory = mock(TransactionBufferSnapshotServiceFactory.class); when(transactionBufferSnapshotServiceFactory.getTxnBufferSnapshotService()) From 3f165617224e65d9eec314139caffddd1bf1562f Mon Sep 17 00:00:00 2001 From: rangao Date: Thu, 2 Mar 2023 21:28:12 +0800 Subject: [PATCH 4/7] address comment --- .../SystemTopicTxnBufferSnapshotService.java | 26 ++++++++----------- ...SingleSnapshotAbortedTxnProcessorImpl.java | 2 +- ...napshotSegmentAbortedTxnProcessorImpl.java | 6 +++-- .../TopicTransactionBufferRecoverTest.java | 4 +-- 4 files changed, 18 insertions(+), 20 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java index abf579db309d6..d5350cdfce7b8 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java @@ -60,7 +60,7 @@ public ReferenceCountedWriter(NamespaceName namespaceName, this.snapshotService = snapshotService; this.future = future; this.future.exceptionally(t -> { - log.error("[{}] Failed to create transaction buffer snapshot writer.", namespaceName, t); + log.error("[{}] Failed to create TB snapshot writer.", namespaceName, t); snapshotService.refCountedWriterMap.remove(namespaceName, this); return null; }); @@ -90,7 +90,7 @@ public synchronized void release() { return null; }); }); - }; + } } } @@ -105,7 +105,7 @@ public SystemTopicTxnBufferSnapshotService(PulsarClient client, EventType system } public CompletableFuture> createReader(TopicName topicName) { - return getTransactionBufferSystemTopicClient(topicName).newReaderAsync(); + return getTransactionBufferSystemTopicClient(topicName.getNamespaceObject()).newReaderAsync(); } public void removeClient(TopicName topicName, SystemTopicClientBase transactionBufferSystemTopicClient) { @@ -115,31 +115,27 @@ public void removeClient(TopicName topicName, SystemTopicClientBase transacti } } - public ReferenceCountedWriter getReferenceWriter(TopicName topicName) { - return refCountedWriterMap.compute(topicName.getNamespaceObject(), (k, v) -> { + public ReferenceCountedWriter getReferenceWriter(NamespaceName namespaceName) { + return refCountedWriterMap.compute(namespaceName, (k, v) -> { if (v != null && v.retain()) { return v; } else { - return new ReferenceCountedWriter<>(topicName.getNamespaceObject(), - getTransactionBufferSystemTopicClient(topicName).newWriterAsync(), this); + return new ReferenceCountedWriter<>(namespaceName, + getTransactionBufferSystemTopicClient(namespaceName).newWriterAsync(), this); } }); } - private SystemTopicClient getTransactionBufferSystemTopicClient(TopicName topicName) { - if (topicName == null) { - throw new RuntimeException(new PulsarClientException - .InvalidTopicNameException("Can't get the tb system topic client due to the topic name is null.")); - } + private SystemTopicClient getTransactionBufferSystemTopicClient(NamespaceName namespaceName) { TopicName systemTopicName = NamespaceEventsSystemTopicFactory - .getSystemTopicName(topicName.getNamespaceObject(), systemTopicType); + .getSystemTopicName(namespaceName, systemTopicType); if (systemTopicName == null) { throw new RuntimeException(new PulsarClientException.InvalidTopicNameException( - "Can't get the tb system topic client for topic " + topicName + "Can't get the TB system topic client for namespace " + namespaceName + " with type " + systemTopicType + ".")); } - return clients.computeIfAbsent(topicName.getNamespaceObject(), + return clients.computeIfAbsent(namespaceName, (v) -> namespaceEventsSystemTopicFactory .createTransactionBufferSystemTopicClient(systemTopicName, this, schemaType)); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java index bb08a7f60309a..5d582d564eadd 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java @@ -58,7 +58,7 @@ public SingleSnapshotAbortedTxnProcessorImpl(PersistentTopic topic) { this.topic = topic; this.takeSnapshotWriter = this.topic.getBrokerService().getPulsar() .getTransactionBufferSnapshotServiceFactory() - .getTxnBufferSnapshotService().getReferenceWriter(TopicName.get(topic.getName())); + .getTxnBufferSnapshotService().getReferenceWriter(TopicName.get(topic.getName()).getNamespaceObject()); this.takeSnapshotWriter.getFuture().exceptionally((ex) -> { log.error("{} Failed to create snapshot writer", topic.getName()); topic.close(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java index 64c9e2b393cae..751c03aff95a9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java @@ -473,7 +473,8 @@ public PersistentWorker(PersistentTopic topic) { this.topic = topic; this.snapshotSegmentsWriter = this.topic.getBrokerService().getPulsar() .getTransactionBufferSnapshotServiceFactory() - .getTxnBufferSnapshotSegmentService().getReferenceWriter(TopicName.get(topic.getName())); + .getTxnBufferSnapshotSegmentService() + .getReferenceWriter(TopicName.get(topic.getName()).getNamespaceObject()); this.snapshotSegmentsWriter.getFuture().exceptionally(ex -> { log.error("{} Failed to create snapshot index writer", topic.getName()); topic.close(); @@ -481,7 +482,8 @@ public PersistentWorker(PersistentTopic topic) { }); this.snapshotIndexWriter = this.topic.getBrokerService().getPulsar() .getTransactionBufferSnapshotServiceFactory() - .getTxnBufferSnapshotIndexService().getReferenceWriter(TopicName.get(topic.getName())); + .getTxnBufferSnapshotIndexService() + .getReferenceWriter(TopicName.get(topic.getName()).getNamespaceObject()); this.snapshotIndexWriter.getFuture().exceptionally((ex) -> { log.error("{} Failed to create snapshot writer", topic.getName()); topic.close(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java index a2de13c0a3de5..2d6622571c033 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java @@ -707,7 +707,7 @@ public void testTransactionBufferIndexSystemTopic() throws Exception { SystemTopicClient.Writer indexesWriter = transactionBufferSnapshotIndexService.getReferenceWriter( - TopicName.get(SNAPSHOT_INDEX)).getFuture().get(); + TopicName.get(SNAPSHOT_INDEX).getNamespaceObject()).getFuture().get(); SystemTopicClient.Reader indexesReader = transactionBufferSnapshotIndexService.createReader(TopicName.get(SNAPSHOT_INDEX)).get(); @@ -769,7 +769,7 @@ public void testTransactionBufferSegmentSystemTopic() throws Exception { SystemTopicClient.Writer segmentWriter = transactionBufferSnapshotSegmentService - .getReferenceWriter(snapshotSegmentTopicName).getFuture().get(); + .getReferenceWriter(snapshotSegmentTopicName.getNamespaceObject()).getFuture().get(); // write two snapshot to snapshot segment topic TransactionBufferSnapshotSegment snapshot = From 5a34e40c17b12bb8e2366306f15e801e37cec84b Mon Sep 17 00:00:00 2001 From: rangao Date: Fri, 3 Mar 2023 11:46:56 +0800 Subject: [PATCH 5/7] fix --- .../org/apache/pulsar/broker/transaction/TransactionTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java index 8e7288c899db6..ad9d1602e5bd8 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java @@ -1689,7 +1689,7 @@ public void testTBSnapshotWriter() throws Exception { admin.topics().createPartitionedTopic(topic, partitionCount); Class clazz = SystemTopicTxnBufferSnapshotService.class; - Field field = clazz.getDeclaredField("writerMap"); + Field field = clazz.getDeclaredField("refCountedWriterMap"); field.setAccessible(true); // inject a failed writer future CompletableFuture> writerFuture = new CompletableFuture<>(); From 6468747aacbc82de8930f02182351d22a003ce4b Mon Sep 17 00:00:00 2001 From: rangao Date: Mon, 13 Mar 2023 20:49:08 +0800 Subject: [PATCH 6/7] fix test --- .../SystemTopicTxnBufferSnapshotService.java | 2 +- .../SegmentAbortedTxnProcessorTest.java | 19 ++++++++++++++----- 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java index d5350cdfce7b8..332d754cf97d2 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java @@ -75,7 +75,7 @@ private synchronized boolean retain() { } public synchronized void release() { - if (this.referenceCount.decrementAndGet() <= 0) { + if (this.referenceCount.decrementAndGet() == 0) { snapshotService.refCountedWriterMap.remove(namespaceName, this); future.thenAccept(writer -> { final String topicName = writer.getSystemTopicClient().getTopicName().toString(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/SegmentAbortedTxnProcessorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/SegmentAbortedTxnProcessorTest.java index ffc059de8e656..c157d7cf8c527 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/SegmentAbortedTxnProcessorTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/SegmentAbortedTxnProcessorTest.java @@ -19,6 +19,8 @@ package org.apache.pulsar.broker.transaction; import static org.testng.Assert.assertTrue; +import static org.testng.Assert.fail; + import java.lang.reflect.Field; import java.util.LinkedList; import java.util.NavigableMap; @@ -34,6 +36,7 @@ import org.apache.commons.lang3.tuple.MutablePair; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.service.BrokerServiceException; +import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService.ReferenceCountedWriter; import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory; import org.apache.pulsar.broker.systopic.SystemTopicClient; @@ -125,6 +128,7 @@ public void testPutAbortedTxnIntoProcessor() throws Exception { newProcessor.trimExpiredAbortedTxns(); //4. Verify the two sealed segment will be deleted. Awaitility.await().untilAsserted(() -> verifyAbortedTxnIDAndSegmentIndex(newProcessor, 11, 4)); + processor.closeAsync().get(5, TimeUnit.SECONDS); } private void waitTaskExecuteCompletely(AbortedTxnProcessor processor) throws Exception { @@ -177,8 +181,11 @@ public void testFuturesCanCompleteWhenItIsCanceled() throws Exception { new MutablePair<>(new CompletableFuture<>(), task))); try { processor.takeAbortedTxnsSnapshot(new PositionImpl(1, 10)).get(2, TimeUnit.SECONDS); + fail("The update index operation should fail."); } catch (Exception e) { Assert.assertTrue(e.getCause() instanceof BrokerServiceException.ServiceUnitNotReadyException); + } finally { + processor.closeAsync().get(5, TimeUnit.SECONDS); } } @@ -200,12 +207,13 @@ public void testClearSnapshotSegments() throws Exception { SnapshotSegmentAbortedTxnProcessorImpl.PersistentWorker worker = (SnapshotSegmentAbortedTxnProcessorImpl.PersistentWorker) field.get(processor); Field indexWriteFutureField = SnapshotSegmentAbortedTxnProcessorImpl - .PersistentWorker.class.getDeclaredField("snapshotIndexWriterFuture"); + .PersistentWorker.class.getDeclaredField("snapshotIndexWriter"); indexWriteFutureField.setAccessible(true); - CompletableFuture> snapshotIndexWriterFuture = - (CompletableFuture>) - indexWriteFutureField.get(worker); - snapshotIndexWriterFuture.get().close(); + ReferenceCountedWriter snapshotIndexWriter = + (ReferenceCountedWriter) indexWriteFutureField.get(worker); + snapshotIndexWriter.release(); + // After release, the writer should be closed, call close method again to make sure the writer was closed. + snapshotIndexWriter.getFuture().get().close(); //3. Try to write a snapshot segment that will fail to update indexes. for (int j = 0; j < SEGMENT_SIZE; j++) { TxnID txnID = new TxnID(0, j); @@ -233,6 +241,7 @@ public void testClearSnapshotSegments() throws Exception { //7. Verify the snapshot segments and index after clearing. verifySnapshotSegmentsSize(PROCESSOR_TOPIC, 0); verifySnapshotSegmentsIndexSize(PROCESSOR_TOPIC, 1); + processor.closeAsync().get(5, TimeUnit.SECONDS); } private void verifySnapshotSegmentsSize(String topic, int size) throws Exception { From bdbc228ac1e5ac157dc537647fa6a12a010b10ad Mon Sep 17 00:00:00 2001 From: rangao Date: Tue, 14 Mar 2023 18:20:49 +0800 Subject: [PATCH 7/7] fix test --- .../pulsar/broker/transaction/TransactionTest.java | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java index ad9d1602e5bd8..c3533e70cf8be 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java @@ -1696,24 +1696,24 @@ public void testTBSnapshotWriter() throws Exception { for (PulsarService pulsarService : pulsarServiceList) { SystemTopicTxnBufferSnapshotService bufferSnapshotService = pulsarService.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotService(); - HashMap writerMap1 = - ((HashMap) field.get(bufferSnapshotService)); + ConcurrentHashMap writerMap1 = + ((ConcurrentHashMap) field.get(bufferSnapshotService)); ReferenceCountedWriter failedCountedWriter = new ReferenceCountedWriter(NamespaceName.get(namespace), writerFuture, bufferSnapshotService); writerMap1.put(NamespaceName.get(namespace), failedCountedWriter); SystemTopicTxnBufferSnapshotService segmentSnapshotService = pulsarService.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotSegmentService(); - HashMap writerMap2 = - ((HashMap) field.get(segmentSnapshotService)); + ConcurrentHashMap writerMap2 = + ((ConcurrentHashMap) field.get(segmentSnapshotService)); ReferenceCountedWriter failedCountedWriter2 = new ReferenceCountedWriter(NamespaceName.get(namespace), writerFuture, segmentSnapshotService); writerMap2.put(NamespaceName.get(namespace), failedCountedWriter2); SystemTopicTxnBufferSnapshotService indexSnapshotService = pulsarService.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotIndexService(); - HashMap writerMap3 = - ((HashMap) field.get(indexSnapshotService)); + ConcurrentHashMap writerMap3 = + ((ConcurrentHashMap) field.get(indexSnapshotService)); ReferenceCountedWriter failedCountedWriter3 = new ReferenceCountedWriter(NamespaceName.get(namespace), writerFuture, indexSnapshotService); writerMap3.put(NamespaceName.get(namespace), failedCountedWriter3);