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 7be599c8c2781..a1b78d89a13eb 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 @@ -72,7 +72,7 @@ protected CompletableFuture> getTransactionBufferSystemTopi } return CompletableFuture.completedFuture(clients.computeIfAbsent(systemTopicName, (v) -> namespaceEventsSystemTopicFactory - .createTransactionBufferSystemTopicClient(topicName.getNamespaceObject(), + .createTransactionBufferSystemTopicClient(systemTopicName, this, schemaType))); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/NamespaceEventsSystemTopicFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/NamespaceEventsSystemTopicFactory.java index 8d30d1d140f5a..f5e6c7748d10b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/NamespaceEventsSystemTopicFactory.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/NamespaceEventsSystemTopicFactory.java @@ -44,12 +44,10 @@ public TopicPoliciesSystemTopicClient createTopicPoliciesSystemTopicClient(Names } public TransactionBufferSnapshotBaseSystemTopicClient createTransactionBufferSystemTopicClient( - NamespaceName namespaceName, SystemTopicTxnBufferSnapshotService + TopicName systemTopicName, SystemTopicTxnBufferSnapshotService systemTopicTxnBufferSnapshotService, Class schemaType) { - TopicName topicName = TopicName.get(TopicDomain.persistent.value(), namespaceName, - SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT); - log.info("Create transaction buffer snapshot client, topicName : {}", topicName.toString()); - return new TransactionBufferSnapshotBaseSystemTopicClient(client, topicName, + log.info("Create transaction buffer snapshot client, topicName : {}", systemTopicName.toString()); + return new TransactionBufferSnapshotBaseSystemTopicClient(client, systemTopicName, systemTopicTxnBufferSnapshotService, schemaType); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/TransactionBufferSnapshotBaseSystemTopicClient.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/TransactionBufferSnapshotBaseSystemTopicClient.java index b18bf552c3004..8efa983a64d73 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/TransactionBufferSnapshotBaseSystemTopicClient.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/TransactionBufferSnapshotBaseSystemTopicClient.java @@ -188,6 +188,7 @@ public SystemTopicClient getSystemTopic() { protected CompletableFuture> newWriterAsyncInternal() { return client.newProducer(Schema.AVRO(schemaType)) .topic(topicName.toString()) + .enableBatching(false) .createAsync().thenApply(producer -> { if (log.isDebugEnabled()) { log.debug("[{}] A new {} writer is created", topicName, schemaType.getName()); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/AbortedTxnProcessor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/AbortedTxnProcessor.java index e436e1df24972..8223aa12b75ae 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/AbortedTxnProcessor.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/AbortedTxnProcessor.java @@ -19,7 +19,6 @@ package org.apache.pulsar.broker.transaction.buffer; import java.util.concurrent.CompletableFuture; -import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.pulsar.client.api.transaction.TxnID; @@ -30,9 +29,9 @@ public interface AbortedTxnProcessor { * After the transaction buffer writes a transaction aborted marker to the topic, * the transaction buffer will put the aborted txnID and the aborted marker position to AbortedTxnProcessor. * @param txnID aborted transaction ID. - * @param position the position of the abort txnID + * @param abortedMarkerPersistentPosition the position of the abort txn marker. */ - void putAbortedTxnAndPosition(TxnID txnID, PositionImpl position); + void putAbortedTxnAndPosition(TxnID txnID, PositionImpl abortedMarkerPersistentPosition); /** * Clean up invalid aborted transactions. @@ -42,10 +41,9 @@ public interface AbortedTxnProcessor { /** * Check whether the transaction ID is an aborted transaction ID. * @param txnID the transaction ID that needs to be checked. - * @param readPosition the read position of the transaction message, can be used to find the segment. * @return a boolean, whether the transaction ID is an aborted transaction ID. */ - boolean checkAbortedTransaction(TxnID txnID, Position readPosition); + boolean checkAbortedTransaction(TxnID txnID); /** * Recover transaction buffer by transaction buffer snapshot. @@ -58,7 +56,7 @@ public interface AbortedTxnProcessor { * Delete the transaction buffer aborted transaction snapshot. * @return a completableFuture. */ - CompletableFuture deleteAbortedTxnSnapshot(); + CompletableFuture clearAbortedTxnSnapshot(); /** * Take aborted transactions snapshot. 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 f8d0d32391233..87161e97512b9 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 @@ -24,7 +24,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import lombok.extern.slf4j.Slf4j; -import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.commons.collections4.map.LinkedMap; @@ -57,11 +56,16 @@ public SingleSnapshotAbortedTxnProcessorImpl(PersistentTopic topic) { this.takeSnapshotWriter = this.topic.getBrokerService().getPulsar() .getTransactionBufferSnapshotServiceFactory() .getTxnBufferSnapshotService().createWriter(TopicName.get(topic.getName())); + this.takeSnapshotWriter.exceptionally((ex) -> { + log.error("{} Failed to create snapshot writer", topic.getName()); + topic.close(); + return null; + }); } @Override - public void putAbortedTxnAndPosition(TxnID abortedTxnId, PositionImpl position) { - aborts.put(abortedTxnId, position); + public void putAbortedTxnAndPosition(TxnID abortedTxnId, PositionImpl abortedMarkerPersistentPosition) { + aborts.put(abortedTxnId, abortedMarkerPersistentPosition); } //In this implementation we clear the invalid aborted txn ID one by one. @@ -78,7 +82,7 @@ public void trimExpiredAbortedTxns() { } @Override - public boolean checkAbortedTransaction(TxnID txnID, Position readPosition) { + public boolean checkAbortedTransaction(TxnID txnID) { return aborts.containsKey(txnID); } @@ -127,14 +131,12 @@ public CompletableFuture recoverFromSnapshot() { } @Override - public CompletableFuture deleteAbortedTxnSnapshot() { + public CompletableFuture clearAbortedTxnSnapshot() { return this.takeSnapshotWriter.thenCompose(writer -> { TransactionBufferSnapshot snapshot = new TransactionBufferSnapshot(); snapshot.setTopicName(topic.getName()); return writer.deleteAsync(snapshot.getTopicName(), snapshot); - }).thenRun(() -> { - log.info("[{}] Successes to delete the aborted transaction snapshot", this.topic); - }); + }).thenRun(() -> log.info("[{}] Successes to delete the aborted transaction snapshot", this.topic)); } @Override 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 new file mode 100644 index 0000000000000..7a9e0e1abedd9 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java @@ -0,0 +1,784 @@ +/* + * 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.broker.transaction.buffer.impl; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import java.util.ArrayList; +import java.util.LinkedList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; +import java.util.function.Supplier; +import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.mledger.AsyncCallbacks; +import org.apache.bookkeeper.mledger.Entry; +import org.apache.bookkeeper.mledger.ManagedLedgerException; +import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; +import org.apache.bookkeeper.mledger.impl.PositionImpl; +import org.apache.bookkeeper.mledger.impl.ReadOnlyManagedLedgerImpl; +import org.apache.commons.collections4.map.LinkedMap; +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.persistent.PersistentTopic; +import org.apache.pulsar.broker.systopic.SystemTopicClient; +import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor; +import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndex; +import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndexes; +import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndexesMetadata; +import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotSegment; +import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TxnIDData; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.transaction.TxnID; +import org.apache.pulsar.client.impl.MessageIdImpl; +import org.apache.pulsar.client.impl.PulsarClientImpl; +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.protocol.Commands; +import org.apache.pulsar.common.util.FutureUtil; + +@Slf4j +public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcessor { + + /** + * Stored the unsealed aborted transaction IDs Whose size is always less than the snapshotSegmentCapacity. + * It will be persistent as a snapshot segment when its size reach the configured capacity. + */ + private LinkedList unsealedTxnIds; + + /** + * The map is used to clear the aborted transaction IDs persistent in the expired ledger. + *

+ * The key PositionImpl {@link PositionImpl} is the persistent position of + * the latest transaction of a segment. + * The value TxnID {@link TxnID} is the latest Transaction ID in a segment. + *

+ * + *

+ * If the position is expired, the processor can get the according latest + * transaction ID in this map. And then the processor can clear all the + * transaction IDs in the aborts {@link SnapshotSegmentAbortedTxnProcessorImpl#aborts} + * that lower than the transaction ID. + * And then the processor can delete the segments persistently according to + * the positions. + *

+ */ + private final LinkedMap segmentIndex = new LinkedMap<>(); + + /** + * This map is used to check whether a transaction is an aborted transaction. + *

+ * The transaction IDs is appended in order, so the processor can delete expired + * transaction IDs according to the latest expired transaction IDs in segmentIndex + * {@link SnapshotSegmentAbortedTxnProcessorImpl#segmentIndex}. + *

+ */ + private final LinkedMap aborts = new LinkedMap<>(); + /** + * This map stores the indexes of the snapshot segment. + *

+ * The key is the persistent position of the marker of the last transaction in the segment. + * The value TransactionBufferSnapshotIndex {@link TransactionBufferSnapshotIndex} is the + * indexes of the snapshot segment. + *

+ */ + private final LinkedMap indexes = new LinkedMap<>(); + + private final PersistentTopic topic; + + private volatile long lastSnapshotTimestamps; + + /** + * The number of the aborted transaction IDs in a segment. + * This is calculated according to the configured memory size. + */ + private final int snapshotSegmentCapacity; + /** + * Responsible for executing the persistent tasks. + *

Including:

+ *

Update segment index.

+ *

Write snapshot segment.

+ *

Delete snapshot segment.

+ *

Clear all snapshot segment.

+ */ + private final PersistentWorker persistentWorker; + + private static final String SNAPSHOT_PREFIX = "multiple-"; + + public SnapshotSegmentAbortedTxnProcessorImpl(PersistentTopic topic) { + this.topic = topic; + this.persistentWorker = new PersistentWorker(topic); + /* + Calculate the segment capital according to its size configuration. +

+ The empty transaction segment size is 5. + Adding an empty linkedList, the size increase to 6. + Add the topic name the size increase to the 7 + topic.getName().length(). + Add the aborted transaction IDs, the size increase to 8 + + topic.getName().length() + 3 * aborted transaction ID size. +

+ */ + this.snapshotSegmentCapacity = (topic.getBrokerService().getPulsar() + .getConfiguration().getTransactionBufferSnapshotSegmentSize() - 8 - topic.getName().length()) / 3; + this.unsealedTxnIds = new LinkedList<>(); + } + + @Override + public void putAbortedTxnAndPosition(TxnID txnID, PositionImpl position) { + unsealedTxnIds.add(txnID); + aborts.put(txnID, txnID); + /* + The size of lastAbortedTxns reaches the configuration of the size of snapshot segment. + Append a task to persistent the segment with the aborted transaction IDs and the latest + transaction mark persistent position passed by param. + */ + if (unsealedTxnIds.size() >= snapshotSegmentCapacity) { + LinkedList abortedSegment = unsealedTxnIds; + segmentIndex.put(position, txnID); + persistentWorker.appendTask(PersistentWorker.OperationType.WriteSegment, + () -> persistentWorker.takeSnapshotSegmentAsync(abortedSegment, position)); + this.unsealedTxnIds = new LinkedList<>(); + } + } + + @Override + public boolean checkAbortedTransaction(TxnID txnID) { + return aborts.containsKey(txnID); + } + + /** + * Check werther the position in segmentIndex {@link SnapshotSegmentAbortedTxnProcessorImpl#segmentIndex} + * is expired. If the position is not exist in the original topic, the according transaction is an invalid + * transaction. And the according segment is invalid, too. The transaction IDs before the transaction ID + * in the aborts are invalid, too. + */ + @Override + public void trimExpiredAbortedTxns() { + //Checking whether there are some segment expired. + List positionsNeedToDelete = new ArrayList<>(); + while (!segmentIndex.isEmpty() && !((ManagedLedgerImpl) topic.getManagedLedger()) + .ledgerExists(segmentIndex.firstKey().getLedgerId())) { + if (log.isDebugEnabled()) { + log.debug("[{}] Topic transaction buffer clear aborted transactions, maxReadPosition : {}", + topic.getName(), segmentIndex.firstKey()); + } + PositionImpl positionNeedToDelete = segmentIndex.firstKey(); + positionsNeedToDelete.add(positionNeedToDelete); + + TxnID theLatestDeletedTxnID = segmentIndex.remove(0); + while (!aborts.firstKey().equals(theLatestDeletedTxnID)) { + aborts.remove(0); + } + aborts.remove(0); + } + //Batch delete the expired segment + if (!positionsNeedToDelete.isEmpty()) { + persistentWorker.appendTask(PersistentWorker.OperationType.DeleteSegment, + () -> persistentWorker.deleteSnapshotSegment(positionsNeedToDelete)); + } + } + + private String buildKey(long sequenceId) { + return SNAPSHOT_PREFIX + sequenceId + "-" + this.topic.getName(); + } + + @Override + public CompletableFuture takeAbortedTxnsSnapshot(PositionImpl maxReadPosition) { + //Store the latest aborted transaction IDs in unsealedTxnIDs and the according the latest max read position. + TransactionBufferSnapshotIndexesMetadata metadata = new TransactionBufferSnapshotIndexesMetadata( + maxReadPosition.getLedgerId(), maxReadPosition.getEntryId(), + convertTypeToTxnIDData(unsealedTxnIds)); + return persistentWorker.appendTask(PersistentWorker.OperationType.UpdateIndex, + () -> persistentWorker.updateSnapshotIndex(metadata)); + } + + @Override + public CompletableFuture recoverFromSnapshot() { + return topic.getBrokerService().getPulsar().getTransactionBufferSnapshotServiceFactory() + .getTxnBufferSnapshotIndexService() + .createReader(TopicName.get(topic.getName())).thenComposeAsync(reader -> { + PositionImpl startReadCursorPosition = null; + TransactionBufferSnapshotIndexes persistentSnapshotIndexes = null; + try { + /* + Read the transaction snapshot segment index. +

+ The processor can get the sequence ID, unsealed transaction IDs, + segment index list and max read position in the snapshot segment index. + Then we can traverse the index list to read all aborted transaction IDs + in segments to aborts. +

+ */ + while (reader.hasMoreEvents()) { + Message message = reader.readNextAsync() + .get(getSystemClientOperationTimeoutMs(), TimeUnit.MILLISECONDS); + if (topic.getName().equals(message.getKey())) { + TransactionBufferSnapshotIndexes transactionBufferSnapshotIndexes = message.getValue(); + if (transactionBufferSnapshotIndexes != null) { + persistentSnapshotIndexes = transactionBufferSnapshotIndexes; + startReadCursorPosition = PositionImpl.get( + transactionBufferSnapshotIndexes.getSnapshot().getMaxReadPositionLedgerId(), + transactionBufferSnapshotIndexes.getSnapshot().getMaxReadPositionEntryId()); + } + } + } + } catch (TimeoutException ex) { + Throwable t = FutureUtil.unwrapCompletionException(ex); + String errorMessage = String.format("[%s] Transaction buffer recover fail by read " + + "transactionBufferSnapshot timeout!", topic.getName()); + log.error(errorMessage, t); + return FutureUtil.failedFuture( + new BrokerServiceException.ServiceUnitNotReadyException(errorMessage, t)); + } catch (Exception ex) { + log.error("[{}] Transaction buffer recover fail when read " + + "transactionBufferSnapshot!", topic.getName(), ex); + return FutureUtil.failedFuture(ex); + } finally { + closeReader(reader); + } + PositionImpl finalStartReadCursorPosition = startReadCursorPosition; + TransactionBufferSnapshotIndexes finalPersistentSnapshotIndexes = persistentSnapshotIndexes; + if (persistentSnapshotIndexes == null) { + return CompletableFuture.completedFuture(null); + } else { + this.unsealedTxnIds = convertTypeToTxnID(persistentSnapshotIndexes + .getSnapshot().getAborts()); + } + //Read snapshot segment to recover aborts. + ArrayList> completableFutures = new ArrayList<>(); + CompletableFuture openManagedLedgerAndHandleSegmentsFuture = new CompletableFuture<>(); + AtomicBoolean hasInvalidIndex = new AtomicBoolean(false); + AsyncCallbacks.OpenReadOnlyManagedLedgerCallback callback = new AsyncCallbacks + .OpenReadOnlyManagedLedgerCallback() { + @Override + public void openReadOnlyManagedLedgerComplete(ReadOnlyManagedLedgerImpl readOnlyManagedLedger, + Object ctx) { + finalPersistentSnapshotIndexes.getIndexList().forEach(index -> { + CompletableFuture handleSegmentFuture = new CompletableFuture<>(); + completableFutures.add(handleSegmentFuture); + readOnlyManagedLedger.asyncReadEntry( + new PositionImpl(index.getSegmentLedgerID(), + index.getSegmentEntryID()), + new AsyncCallbacks.ReadEntryCallback() { + @Override + public void readEntryComplete(Entry entry, Object ctx) { + handleSnapshotSegmentEntry(entry); + indexes.put(new PositionImpl( + index.abortedMarkLedgerID, + index.abortedMarkEntryID), + index); + entry.release(); + handleSegmentFuture.complete(null); + } + + @Override + public void readEntryFailed(ManagedLedgerException exception, Object ctx) { + /* + The logic flow of deleting expired segment is: +

+ 1. delete segment + 2. update segment index +

+ If the worker delete segment successfully + but failed to update segment index, + the segment can not be read according to the index. + We update index again if there are invalid indexes. + */ + if (((ManagedLedgerImpl) topic.getManagedLedger()) + .ledgerExists(index.getAbortedMarkLedgerID())) { + log.error("[{}] Failed to read snapshot segment [{}:{}]", + topic.getName(), index.segmentLedgerID, + index.segmentEntryID, exception); + handleSegmentFuture.completeExceptionally(exception); + } else { + hasInvalidIndex.set(true); + } + } + }, null); + }); + openManagedLedgerAndHandleSegmentsFuture.complete(null); + } + + @Override + public void openReadOnlyManagedLedgerFailed(ManagedLedgerException exception, Object ctx) { + log.error("[{}] Failed to open readOnly managed ledger", topic, exception); + openManagedLedgerAndHandleSegmentsFuture.completeExceptionally(exception); + } + }; + + TopicName snapshotSegmentTopicName = TopicName.get(TopicDomain.persistent.toString(), + TopicName.get(topic.getName()).getNamespaceObject(), + SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT_SEGMENTS); + this.topic.getBrokerService().getPulsar().getManagedLedgerFactory() + .asyncOpenReadOnlyManagedLedger(snapshotSegmentTopicName + .getPersistenceNamingEncoding(), callback, + topic.getManagedLedger().getConfig(), + null); + /* + Wait the processor recover completely and then allow TB + to recover the messages after the startReadCursorPosition. + */ + return openManagedLedgerAndHandleSegmentsFuture + .thenCompose((ignore) -> FutureUtil.waitForAll(completableFutures)) + .thenCompose((i) -> { + /* + Update the snapshot segment index if there exist invalid indexes. + */ + if (hasInvalidIndex.get()) { + persistentWorker.appendTask(PersistentWorker.OperationType.UpdateIndex, + () -> persistentWorker.updateSnapshotIndex( + finalPersistentSnapshotIndexes.getSnapshot())); + } + /* + If there is no segment index, the persistent worker will write segment begin from 0. + */ + if (indexes.size() != 0) { + persistentWorker.sequenceID.set(indexes.get(indexes.lastKey()).sequenceID + 1); + } + /* + Append the aborted txn IDs in the index metadata + can keep the order of the aborted txn in the aborts. + So that we can trim the expired snapshot segment in aborts + according to the latest transaction IDs in the segmentIndex. + */ + unsealedTxnIds.forEach(txnID -> aborts.put(txnID, txnID)); + return CompletableFuture.completedFuture(finalStartReadCursorPosition); + }).exceptionally(ex -> { + log.error("[{}] Failed to recover snapshot segment", this.topic.getName(), ex); + return null; + }); + + }, topic.getBrokerService().getPulsar().getTransactionExecutorProvider() + .getExecutor(this)); + } + + @Override + public CompletableFuture clearAbortedTxnSnapshot() { + return persistentWorker.appendTask(PersistentWorker.OperationType.Clear, + persistentWorker::clearSnapshotSegmentAndIndexes); + } + + @Override + public long getLastSnapshotTimestamps() { + return this.lastSnapshotTimestamps; + } + + @Override + public CompletableFuture closeAsync() { + return persistentWorker.closeAsync(); + } + + private void handleSnapshotSegmentEntry(Entry entry) { + //decode snapshot from entry + ByteBuf headersAndPayload = entry.getDataBuffer(); + //skip metadata + Commands.parseMessageMetadata(headersAndPayload); + TransactionBufferSnapshotSegment snapshotSegment = Schema.AVRO(TransactionBufferSnapshotSegment.class) + .decode(Unpooled.wrappedBuffer(headersAndPayload).nioBuffer()); + + TxnIDData lastTxn = snapshotSegment.getAborts().get(snapshotSegment.getAborts().size() - 1); + segmentIndex.put(new PositionImpl(snapshotSegment.getPersistentPositionLedgerId(), + snapshotSegment.getPersistentPositionEntryId()), + new TxnID(lastTxn.getMostSigBits(), lastTxn.getLeastSigBits())); + convertTypeToTxnID(snapshotSegment.getAborts()).forEach(txnID -> aborts.put(txnID, txnID)); + } + + private long getSystemClientOperationTimeoutMs() throws Exception { + PulsarClientImpl pulsarClient = (PulsarClientImpl) topic.getBrokerService().getPulsar().getClient(); + return pulsarClient.getConfiguration().getOperationTimeoutMs(); + } + + private void closeReader(SystemTopicClient.Reader reader) { + reader.closeAsync().exceptionally(e -> { + log.error("[{}]Transaction buffer snapshot reader close error!", topic.getName(), e); + return null; + }); + } + + /** + * The PersistentWorker be responsible for executing the persistent tasks, including: + *

+ * 1. Write snapshot segment --- Encapsulate a sealed snapshot segment and persistent it. + * 2. Delete snapshot segment --- Evict expired snapshot segments. + * 3. Update snapshot indexes --- Update snapshot indexes after writing or deleting snapshot segment + * or update snapshot indexes metadata regularly. + * 4. Clear all snapshot segments and indexes. --- Executed when deleting this topic. + *

+ * * Task 1 and task 2 will be put into a task queue. The tasks in the queue will be executed in order. + * * If the task queue is empty, task 3 will be executed immediately when it is appended to the worker. + * Else, the worker will try to execute the tasks in the task queue. + * * When task 4 was appended into worker, the worker will change the operation state to closed + * and cancel all tasks in the task queue. finally, execute the task 4 (clear task). + * If there are race conditions, throw an Exception to let users try again. + */ + public class PersistentWorker { + protected final AtomicLong sequenceID = new AtomicLong(0); + + private final PersistentTopic topic; + + //Persistent snapshot segment and index at the single thread. + private final CompletableFuture> + snapshotSegmentsWriterFuture; + private final CompletableFuture> + snapshotIndexWriterFuture; + + private enum OperationState { + None, + Operating, + Closed + } + private static final AtomicReferenceFieldUpdater + STATE_UPDATER = AtomicReferenceFieldUpdater.newUpdater(PersistentWorker.class, + PersistentWorker.OperationState.class, "operationState"); + + public enum OperationType { + UpdateIndex, + WriteSegment, + DeleteSegment, + Clear + } + + private volatile OperationState operationState = OperationState.None; + + ConcurrentLinkedDeque, + Supplier>>>> taskQueue = new ConcurrentLinkedDeque<>(); + + public PersistentWorker(PersistentTopic topic) { + this.topic = topic; + this.snapshotSegmentsWriterFuture = this.topic.getBrokerService().getPulsar() + .getTransactionBufferSnapshotServiceFactory() + .getTxnBufferSnapshotSegmentService().createWriter(TopicName.get(topic.getName())); + this.snapshotSegmentsWriterFuture.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) -> { + log.error("{} Failed to create snapshot writer", topic.getName()); + topic.close(); + return null; + }); + } + + public CompletableFuture appendTask(OperationType operationType, + Supplier> task) { + CompletableFuture taskExecutedResult = new CompletableFuture<>(); + switch (operationType) { + case UpdateIndex -> { + /* + The update index operation can be canceled when the task queue is not empty, + so it should be executed immediately instead of appending to the task queue. + If the taskQueue is not empty, the worker will execute the tasks in the queue. + */ + if (!taskQueue.isEmpty()) { + executeTask(); + return cancelUpdateIndexTask(); + } else if (STATE_UPDATER.compareAndSet(this, OperationState.None, OperationState.Operating)) { + return task.get().whenComplete((ignore, throwable) -> { + if (throwable != null && log.isDebugEnabled()) { + log.debug("[{}] Failed to update index snapshot", topic.getName(), throwable); + } + STATE_UPDATER.compareAndSet(this, OperationState.Operating, OperationState.None); + }); + } else { + return cancelUpdateIndexTask(); + } + } + /* + Only the operations of WriteSegment and DeleteSegment will be appended into the taskQueue. + The operation will be canceled when the worker is close which means the topic is deleted. + */ + case WriteSegment, DeleteSegment -> { + if (!STATE_UPDATER.get(this).equals(OperationState.Closed)) { + taskQueue.add(new MutablePair<>(operationType, new MutablePair<>(taskExecutedResult, task))); + executeTask(); + return taskExecutedResult; + } else { + return CompletableFuture.completedFuture(null); + } + } + case Clear -> { + /* + Do not clear the snapshots if the topic is used. + If the users want to delete a topic, they should stop the usage of the topic. + */ + if (STATE_UPDATER.compareAndSet(this, OperationState.None, OperationState.Closed)) { + taskQueue.forEach(pair -> + pair.getRight().getRight().get().completeExceptionally( + new BrokerServiceException.ServiceUnitNotReadyException( + String.format("Cancel the operation [%s] due to the" + + " transaction buffer of the topic[%s] already closed", + pair.getLeft().name(), this.topic.getName())))); + taskQueue.clear(); + /* + The task of clear all snapshot segments and indexes is executed immediately. + */ + return task.get(); + } else { + return FutureUtil.failedFuture( + new BrokerServiceException.NotAllowedException( + String.format("Failed to clear the snapshot of topic [%s] due to " + + "the topic is used. Please stop the using of the topic " + + "and try it again", this.topic.getName()))); + } + } + default -> { + return FutureUtil.failedFuture(new BrokerServiceException + .NotAllowedException(String.format("Th operation [%s] is unsupported", + operationType.name()))); + } + } + } + + private CompletableFuture cancelUpdateIndexTask() { + if (log.isDebugEnabled()) { + log.debug("The operation of updating index is canceled due there is other operation executing"); + } + return FutureUtil.failedFuture(new BrokerServiceException + .ServiceUnitNotReadyException("The operation of updating index is canceled")); + } + + private void executeTask() { + if (taskQueue.isEmpty()) { + return; + } + if (STATE_UPDATER.compareAndSet(this, OperationState.None, OperationState.Operating)) { + //Double-check. Avoid NoSuchElementException due to the first task is completed by other thread. + if (taskQueue.isEmpty()) { + return; + } + Pair, Supplier>>> firstTask = + taskQueue.getFirst(); + firstTask.getValue().getRight().get().whenComplete((ignore, throwable) -> { + if (throwable != null) { + if (log.isDebugEnabled()) { + log.debug("[{}] Failed to do operation do operation of [{}]", + topic.getName(), firstTask.getKey().name(), throwable); + } + //Do not execute the tasks in the task queue until the next task is appended to the task queue. + firstTask.getRight().getKey().completeExceptionally(throwable); + } else { + firstTask.getRight().getKey().complete(null); + taskQueue.removeFirst(); + //Execute the next task in the other thread. + topic.getBrokerService().getPulsar().getTransactionExecutorProvider() + .getExecutor(this).submit(this::executeTask); + } + STATE_UPDATER.compareAndSet(this, OperationState.Operating, + OperationState.None); + }); + } + } + + private CompletableFuture takeSnapshotSegmentAsync(LinkedList sealedAbortedTxnIdSegment, + PositionImpl abortedMarkerPersistentPosition) { + CompletableFuture res = writeSnapshotSegmentAsync(sealedAbortedTxnIdSegment, + abortedMarkerPersistentPosition).thenRun(() -> { + if (log.isDebugEnabled()) { + log.debug("Successes to take snapshot segment [{}] at maxReadPosition [{}] " + + "for the topic [{}], and the size of the segment is [{}]", + this.sequenceID, abortedMarkerPersistentPosition, topic.getName(), + sealedAbortedTxnIdSegment.size()); + } + this.sequenceID.getAndIncrement(); + }); + res.exceptionally(e -> { + //Just log the error, and the processor will try to take snapshot again when the transactionBuffer + //append aborted txn next time. + log.error("Failed to take snapshot segment [{}] at maxReadPosition [{}] " + + "for the topic [{}], and the size of the segment is [{}]", + this.sequenceID, abortedMarkerPersistentPosition, topic.getName(), + sealedAbortedTxnIdSegment.size(), e); + return null; + }); + return res; + } + + private CompletableFuture writeSnapshotSegmentAsync(LinkedList segment, + PositionImpl abortedMarkerPersistentPosition) { + TransactionBufferSnapshotSegment transactionBufferSnapshotSegment = new TransactionBufferSnapshotSegment(); + transactionBufferSnapshotSegment.setAborts(convertTypeToTxnIDData(segment)); + transactionBufferSnapshotSegment.setTopicName(this.topic.getName()); + transactionBufferSnapshotSegment.setPersistentPositionEntryId(abortedMarkerPersistentPosition.getEntryId()); + transactionBufferSnapshotSegment.setPersistentPositionLedgerId( + abortedMarkerPersistentPosition.getLedgerId()); + + return snapshotSegmentsWriterFuture.thenCompose(segmentWriter -> { + transactionBufferSnapshotSegment.setSequenceId(this.sequenceID.get()); + return segmentWriter.writeAsync(buildKey(this.sequenceID.get()), transactionBufferSnapshotSegment); + }).thenCompose((messageId) -> { + //Build index for this segment + TransactionBufferSnapshotIndex index = new TransactionBufferSnapshotIndex(); + index.setSequenceID(transactionBufferSnapshotSegment.getSequenceId()); + index.setAbortedMarkLedgerID(abortedMarkerPersistentPosition.getLedgerId()); + index.setAbortedMarkEntryID(abortedMarkerPersistentPosition.getEntryId()); + index.setSegmentLedgerID(((MessageIdImpl) messageId).getLedgerId()); + index.setSegmentEntryID(((MessageIdImpl) messageId).getEntryId()); + + indexes.put(abortedMarkerPersistentPosition, index); + //update snapshot segment index. + //If the index can not be written successfully, the snapshot segment wil be overwritten + //when the processor writes snapshot segment next time. + //And if the task is not the newest in the queue, it is no need to update the index. + return updateIndexWhenExecuteTheLatestTask(); + }); + } + + private CompletionStage updateIndexWhenExecuteTheLatestTask() { + PositionImpl maxReadPosition = topic.getMaxReadPosition(); + List aborts = convertTypeToTxnIDData(unsealedTxnIds); + if (taskQueue.size() != 1) { + return CompletableFuture.completedFuture(null); + } else { + return updateSnapshotIndex(new TransactionBufferSnapshotIndexesMetadata( + maxReadPosition.getLedgerId(), maxReadPosition.getEntryId(), aborts)); + } + } + + // update index after delete all segment. + private CompletableFuture deleteSnapshotSegment(List positionNeedToDeletes) { + List> results = new ArrayList<>(); + for (PositionImpl positionNeedToDelete : positionNeedToDeletes) { + long sequenceIdNeedToDelete = indexes.get(positionNeedToDelete).getSequenceID(); + CompletableFuture res = snapshotSegmentsWriterFuture + .thenCompose(writer -> writer.deleteAsync(buildKey(sequenceIdNeedToDelete), null)) + .thenCompose(messageId -> { + if (log.isDebugEnabled()) { + log.debug("[{}] Successes to delete the snapshot segment, " + + "whose sequenceId is [{}] and maxReadPosition is [{}]", + this.topic.getName(), this.sequenceID, positionNeedToDelete); + } + //The index may fail to update but the processor will check + //whether the snapshot segment is null, and update the index when recovering. + //And if the task is not the newest in the queue, it is no need to update the index. + indexes.remove(positionNeedToDelete); + return updateIndexWhenExecuteTheLatestTask(); + }); + res.exceptionally(e -> { + log.warn("[{}] Failed to delete the snapshot segment, " + + "whose sequenceId is [{}] and maxReadPosition is [{}]", + this.topic.getName(), this.sequenceID, positionNeedToDelete, e); + return null; + }); + results.add(res); + } + return FutureUtil.waitForAll(results); + } + + private CompletableFuture updateSnapshotIndex(TransactionBufferSnapshotIndexesMetadata snapshotSegment) { + TransactionBufferSnapshotIndexes snapshotIndexes = new TransactionBufferSnapshotIndexes(); + CompletableFuture res = snapshotIndexWriterFuture + .thenCompose((indexesWriter) -> { + snapshotIndexes.setIndexList(indexes.values().stream().toList()); + snapshotIndexes.setSnapshot(snapshotSegment); + snapshotIndexes.setTopicName(topic.getName()); + return indexesWriter.writeAsync(topic.getName(), snapshotIndexes) + .thenCompose(messageId -> CompletableFuture.completedFuture(null)); + }); + res.thenRun(() -> lastSnapshotTimestamps = System.currentTimeMillis()).exceptionally(e -> { + log.error("[{}] Failed to update snapshot segment index", snapshotIndexes.getTopicName(), e); + return null; + }); + return res; + } + + private CompletableFuture clearSnapshotSegmentAndIndexes() { + CompletableFuture res = persistentWorker.clearAllSnapshotSegments() + .thenCompose((ignore) -> snapshotIndexWriterFuture + .thenCompose(indexesWriter -> indexesWriter.writeAsync(topic.getName(), null))) + .thenRun(() -> + log.debug("Successes to clear the snapshot segment and indexes for the topic [{}]", + topic.getName())); + res.exceptionally(e -> { + log.error("Failed to clear the snapshot segment and indexes for the topic [{}]", + topic.getName(), e); + return null; + }); + return res; + } + + /** + * Because the operation of writing segment and index is not atomic, + * we cannot use segment index to clear the snapshot segments. + * If we use segment index to clear snapshot segments, there will case dirty data in the below case: + *

+ * 1. Write snapshot segment 1, 2, 3, update index (1, 2, 3) + * 2. Write snapshot 4, failing to update index + * 3. Trim expired snapshot segment 1, 2, 3, update index (empty) + * 4. Write snapshot segment 1, 2, update index (1, 2) + * 5. Delete topic, clear all snapshot segment (segment1. segment2). + * Segment 3 and segment 4 can not be cleared until this namespace being deleted. + *

+ */ + private CompletableFuture clearAllSnapshotSegments() { + return topic.getBrokerService().getPulsar().getTransactionBufferSnapshotServiceFactory() + .getTxnBufferSnapshotSegmentService() + .createReader(TopicName.get(topic.getName())).thenComposeAsync(reader -> { + try { + while (reader.hasMoreEvents()) { + Message message = reader.readNextAsync() + .get(getSystemClientOperationTimeoutMs(), TimeUnit.MILLISECONDS); + if (topic.getName().equals(message.getValue().getTopicName())) { + snapshotSegmentsWriterFuture.get().write(message.getKey(), null); + } + } + return CompletableFuture.completedFuture(null); + } catch (Exception ex) { + log.error("[{}] Transaction buffer clear snapshot segments fail!", topic.getName(), ex); + return FutureUtil.failedFuture(ex); + } finally { + closeReader(reader); + } + }); + } + + + CompletableFuture closeAsync() { + return CompletableFuture.allOf( + this.snapshotIndexWriterFuture.thenCompose(SystemTopicClient.Writer::closeAsync), + this.snapshotSegmentsWriterFuture.thenCompose(SystemTopicClient.Writer::closeAsync)); + } + } + + private LinkedList convertTypeToTxnID(List snapshotSegment) { + LinkedList abortedTxns = new LinkedList<>(); + snapshotSegment.forEach(txnIDData -> + abortedTxns.add(new TxnID(txnIDData.getMostSigBits(), txnIDData.getLeastSigBits()))); + return abortedTxns; + } + + private List convertTypeToTxnIDData(List abortedTxns) { + List segment = new LinkedList<>(); + abortedTxns.forEach(txnID -> segment.add(new TxnIDData(txnID.getMostSigBits(), txnID.getLeastSigBits()))); + return segment; + } + +} \ No newline at end of file diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java index f3bf4f95923cd..89a8e95afba1f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java @@ -110,7 +110,11 @@ public TopicTransactionBuffer(PersistentTopic topic) { this.takeSnapshotIntervalTime = topic.getBrokerService().getPulsar() .getConfiguration().getTransactionBufferSnapshotMinTimeInMillis(); this.maxReadPosition = (PositionImpl) topic.getManagedLedger().getLastConfirmedEntry(); - this.snapshotAbortedTxnProcessor = new SingleSnapshotAbortedTxnProcessorImpl(topic); + if (topic.getBrokerService().getPulsar().getConfiguration().isTransactionBufferSegmentedSnapshotEnabled()) { + snapshotAbortedTxnProcessor = new SnapshotSegmentAbortedTxnProcessorImpl(topic); + } else { + snapshotAbortedTxnProcessor = new SingleSnapshotAbortedTxnProcessorImpl(topic); + } this.recover(); } @@ -275,7 +279,7 @@ public void addFailed(ManagedLedgerException exception, Object ctx) { private void handleTransactionMessage(TxnID txnId, Position position) { if (!ongoingTxns.containsKey(txnId) && !this.snapshotAbortedTxnProcessor - .checkAbortedTransaction(txnId, position)) { + .checkAbortedTransaction(txnId)) { ongoingTxns.put(txnId, (PositionImpl) position); PositionImpl firstPosition = ongoingTxns.get(ongoingTxns.firstKey()); //max read position is less than first ongoing transaction message position, so entryId -1 @@ -349,8 +353,8 @@ public CompletableFuture abortTxn(TxnID txnID, long lowWaterMark) { @Override public void addComplete(Position position, ByteBuf entryData, Object ctx) { synchronized (TopicTransactionBuffer.this) { + snapshotAbortedTxnProcessor.putAbortedTxnAndPosition(txnID, (PositionImpl) position); updateMaxReadPosition(txnID); - snapshotAbortedTxnProcessor.putAbortedTxnAndPosition(txnID, maxReadPosition); snapshotAbortedTxnProcessor.trimExpiredAbortedTxns(); takeSnapshotByChangeTimes(); } @@ -455,7 +459,7 @@ public CompletableFuture purgeTxns(List dataLedgers) { @Override public CompletableFuture clearSnapshot() { - return snapshotAbortedTxnProcessor.deleteAbortedTxnSnapshot(); + return snapshotAbortedTxnProcessor.clearAbortedTxnSnapshot(); } @Override @@ -466,7 +470,7 @@ public CompletableFuture closeAsync() { @Override public boolean isTxnAborted(TxnID txnID, PositionImpl readPosition) { - return snapshotAbortedTxnProcessor.checkAbortedTransaction(txnID, readPosition); + return snapshotAbortedTxnProcessor.checkAbortedTransaction(txnID); } @Override diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/metadata/v2/TransactionBufferSnapshotIndex.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/metadata/v2/TransactionBufferSnapshotIndex.java index 118472397adda..b86edc845c1e5 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/metadata/v2/TransactionBufferSnapshotIndex.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/metadata/v2/TransactionBufferSnapshotIndex.java @@ -29,8 +29,21 @@ @NoArgsConstructor public class TransactionBufferSnapshotIndex { public long sequenceID; - public long maxReadPositionLedgerID; - public long maxReadPositionEntryID; - public long persistentPositionLedgerID; - public long persistentPositionEntryID; + /** + * Location(ledger id of position) of a transaction marker in the origin topic. + */ + public long abortedMarkLedgerID; + + /** + * Location(entry id of position) of a transaction marker in the origin topic. + */ + public long abortedMarkEntryID; + /** + * Location(ledger id of position) of a segment data in the system topic __transaction_buffer_snapshot_segments. + */ + public long segmentLedgerID; + /** + * Location(entry id of position) of a segment data in the system topic __transaction_buffer_snapshot_segments. + */ + public long segmentEntryID; } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/metadata/v2/TransactionBufferSnapshotIndexesMetadata.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/metadata/v2/TransactionBufferSnapshotIndexesMetadata.java index 9a468d250bbbf..f9c28a818f8b8 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/metadata/v2/TransactionBufferSnapshotIndexesMetadata.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/metadata/v2/TransactionBufferSnapshotIndexesMetadata.java @@ -18,7 +18,7 @@ */ package org.apache.pulsar.broker.transaction.buffer.metadata.v2; -import java.util.Set; +import java.util.List; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; @@ -29,5 +29,5 @@ public class TransactionBufferSnapshotIndexesMetadata { private long maxReadPositionLedgerId; private long maxReadPositionEntryId; - private Set aborts; + private List aborts; } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/metadata/v2/TransactionBufferSnapshotSegment.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/metadata/v2/TransactionBufferSnapshotSegment.java index 77bb546880dbe..7ca828cc3e64f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/metadata/v2/TransactionBufferSnapshotSegment.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/metadata/v2/TransactionBufferSnapshotSegment.java @@ -29,7 +29,7 @@ public class TransactionBufferSnapshotSegment { private String topicName; private long sequenceId; - private long maxReadPositionLedgerId; - private long maxReadPositionEntryId; + private long persistentPositionLedgerId; + private long persistentPositionEntryId; private List aborts; } 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 new file mode 100644 index 0000000000000..ffc059de8e656 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/SegmentAbortedTxnProcessorTest.java @@ -0,0 +1,280 @@ +/* + * 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.broker.transaction; + +import static org.testng.Assert.assertTrue; +import java.lang.reflect.Field; +import java.util.LinkedList; +import java.util.NavigableMap; +import java.util.Queue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; +import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; +import org.apache.bookkeeper.mledger.impl.PositionImpl; +import org.apache.bookkeeper.mledger.proto.MLDataFormats; +import org.apache.commons.collections4.map.LinkedMap; +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.persistent.PersistentTopic; +import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory; +import org.apache.pulsar.broker.systopic.SystemTopicClient; +import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor; +import org.apache.pulsar.broker.transaction.buffer.impl.SnapshotSegmentAbortedTxnProcessorImpl; +import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndexes; +import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotSegment; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.transaction.TxnID; +import org.apache.pulsar.common.events.EventType; +import org.apache.pulsar.common.naming.TopicName; +import org.testcontainers.shaded.org.awaitility.Awaitility; +import org.testng.Assert; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +@Slf4j +public class SegmentAbortedTxnProcessorTest extends TransactionTestBase { + + private static final String PROCESSOR_TOPIC = "persistent://" + NAMESPACE1 + "/abortedTxnProcessor"; + private static final int SEGMENT_SIZE = 5; + private PulsarService pulsarService = null; + + @Override + @BeforeClass + protected void setup() throws Exception { + setUpBase(1, 1, PROCESSOR_TOPIC, 0); + this.pulsarService = getPulsarServiceList().get(0); + this.pulsarService.getConfig().setTransactionBufferSegmentedSnapshotEnabled(true); + this.pulsarService.getConfig().setTransactionBufferSnapshotSegmentSize(8 + PROCESSOR_TOPIC.length() + + SEGMENT_SIZE * 3); + } + + @Override + @AfterClass + protected void cleanup() throws Exception { + super.internalCleanup(); + } + + /** + * Test api: + * 1. putAbortedTxnAndPosition + * 2. checkAbortedTransaction + * 3. takeAbortedTxnsSnapshot + * 4. recoverFromSnapshot + * 5. trimExpiredAbortedTxns + * @throws Exception + */ + @Test + public void testPutAbortedTxnIntoProcessor() throws Exception { + PersistentTopic persistentTopic = (PersistentTopic) pulsarService.getBrokerService() + .getTopic(PROCESSOR_TOPIC, false).get().get(); + AbortedTxnProcessor processor = new SnapshotSegmentAbortedTxnProcessorImpl(persistentTopic); + //1. prepare test data. + //1.1 Put 10 aborted txn IDs to persistent two sealed segments. + for (int i = 0; i < 10; i++) { + TxnID txnID = new TxnID(0, i); + PositionImpl position = new PositionImpl(0, i); + processor.putAbortedTxnAndPosition(txnID, position); + } + //1.2 Put 4 aborted txn IDs into the unsealed segment. + for (int i = 10; i < 14; i++) { + TxnID txnID = new TxnID(0, i); + PositionImpl position = new PositionImpl(0, i); + processor.putAbortedTxnAndPosition(txnID, position); + } + //1.3 Verify the common data flow + verifyAbortedTxnIDAndSegmentIndex(processor, 0, 14); + //2. Take the latest snapshot and verify recover from snapshot + AbortedTxnProcessor newProcessor = new SnapshotSegmentAbortedTxnProcessorImpl(persistentTopic); + PositionImpl maxReadPosition = new PositionImpl(0, 14); + //2.1 Avoid update operation being canceled. + waitTaskExecuteCompletely(processor); + //2.2 take the latest snapshot + processor.takeAbortedTxnsSnapshot(maxReadPosition).get(); + newProcessor.recoverFromSnapshot().get(); + //Verify the recovery data flow + verifyAbortedTxnIDAndSegmentIndex(newProcessor, 0, 14); + //3. Delete the ledgers and then verify the date. + Field ledgersField = ManagedLedgerImpl.class.getDeclaredField("ledgers"); + ledgersField.setAccessible(true); + NavigableMap ledgers = + (NavigableMap) + ledgersField.get(persistentTopic.getManagedLedger()); + ledgers.forEach((k, v) -> { + ledgers.remove(k); + }); + newProcessor.trimExpiredAbortedTxns(); + //4. Verify the two sealed segment will be deleted. + Awaitility.await().untilAsserted(() -> verifyAbortedTxnIDAndSegmentIndex(newProcessor, 11, 4)); + } + + private void waitTaskExecuteCompletely(AbortedTxnProcessor processor) throws Exception { + Field workerField = SnapshotSegmentAbortedTxnProcessorImpl.class.getDeclaredField("persistentWorker"); + workerField.setAccessible(true); + SnapshotSegmentAbortedTxnProcessorImpl.PersistentWorker persistentWorker = + (SnapshotSegmentAbortedTxnProcessorImpl.PersistentWorker) workerField.get(processor); + Field taskQueueField = SnapshotSegmentAbortedTxnProcessorImpl.PersistentWorker.class + .getDeclaredField("taskQueue"); + taskQueueField.setAccessible(true); + Queue queue = (Queue) taskQueueField.get(persistentWorker); + Awaitility.await().untilAsserted(() -> Assert.assertEquals(queue.size(), 0)); + } + + private void verifyAbortedTxnIDAndSegmentIndex(AbortedTxnProcessor processor, int begin, int txnIdSize) + throws Exception { + //Verify the checking of the aborted txn IDs + for (int i = begin; i < txnIdSize; i++) { + Assert.assertTrue(processor.checkAbortedTransaction(new TxnID(0, i))); + } + //Verify there are 2 sealed segment and the unsealed segment size is 4. + Field unsealedSegmentField = SnapshotSegmentAbortedTxnProcessorImpl.class + .getDeclaredField("unsealedTxnIds"); + Field indexField = SnapshotSegmentAbortedTxnProcessorImpl.class + .getDeclaredField("segmentIndex"); + unsealedSegmentField.setAccessible(true); + indexField.setAccessible(true); + LinkedList unsealedSegment = (LinkedList) unsealedSegmentField.get(processor); + LinkedMap indexes = (LinkedMap) indexField.get(processor); + Assert.assertEquals(unsealedSegment.size(), txnIdSize % SEGMENT_SIZE); + Assert.assertEquals(indexes.size(), txnIdSize / SEGMENT_SIZE); + } + + // Verify the update index future can be completed when the queue has other tasks. + @Test + public void testFuturesCanCompleteWhenItIsCanceled() throws Exception { + PersistentTopic persistentTopic = (PersistentTopic) pulsarService.getBrokerService() + .getTopic(PROCESSOR_TOPIC, false).get().get(); + AbortedTxnProcessor processor = new SnapshotSegmentAbortedTxnProcessorImpl(persistentTopic); + Field workerField = SnapshotSegmentAbortedTxnProcessorImpl.class.getDeclaredField("persistentWorker"); + workerField.setAccessible(true); + SnapshotSegmentAbortedTxnProcessorImpl.PersistentWorker persistentWorker = + (SnapshotSegmentAbortedTxnProcessorImpl.PersistentWorker) workerField.get(processor); + Field taskQueueField = SnapshotSegmentAbortedTxnProcessorImpl.PersistentWorker.class + .getDeclaredField("taskQueue"); + taskQueueField.setAccessible(true); + Supplier task = CompletableFuture::new; + Queue queue = (Queue) taskQueueField.get(persistentWorker); + queue.add(new MutablePair<>(SnapshotSegmentAbortedTxnProcessorImpl.PersistentWorker.OperationType.WriteSegment, + new MutablePair<>(new CompletableFuture<>(), task))); + try { + processor.takeAbortedTxnsSnapshot(new PositionImpl(1, 10)).get(2, TimeUnit.SECONDS); + } catch (Exception e) { + Assert.assertTrue(e.getCause() instanceof BrokerServiceException.ServiceUnitNotReadyException); + } + } + + @Test + public void testClearSnapshotSegments() throws Exception { + PersistentTopic persistentTopic = (PersistentTopic) pulsarService.getBrokerService() + .getTopic(PROCESSOR_TOPIC, false).get().get(); + AbortedTxnProcessor processor = new SnapshotSegmentAbortedTxnProcessorImpl(persistentTopic); + //1. Write two snapshot segment. + for (int j = 0; j < SEGMENT_SIZE * 2; j++) { + TxnID txnID = new TxnID(0, j); + PositionImpl position = new PositionImpl(0, j); + processor.putAbortedTxnAndPosition(txnID, position); + } + Awaitility.await().untilAsserted(() -> verifySnapshotSegmentsSize(PROCESSOR_TOPIC, 2)); + //2. Close index writer, making the index can not be updated. + Field field = SnapshotSegmentAbortedTxnProcessorImpl.class.getDeclaredField("persistentWorker"); + field.setAccessible(true); + SnapshotSegmentAbortedTxnProcessorImpl.PersistentWorker worker = + (SnapshotSegmentAbortedTxnProcessorImpl.PersistentWorker) field.get(processor); + Field indexWriteFutureField = SnapshotSegmentAbortedTxnProcessorImpl + .PersistentWorker.class.getDeclaredField("snapshotIndexWriterFuture"); + indexWriteFutureField.setAccessible(true); + CompletableFuture> snapshotIndexWriterFuture = + (CompletableFuture>) + indexWriteFutureField.get(worker); + snapshotIndexWriterFuture.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); + PositionImpl position = new PositionImpl(0, j); + processor.putAbortedTxnAndPosition(txnID, position); + } + //4. Wait writing segment completed. + Awaitility.await().untilAsserted(() -> verifySnapshotSegmentsSize(PROCESSOR_TOPIC, 3)); + //5. Clear all the snapshot segments and indexes. + try { + processor.clearAbortedTxnSnapshot().get(); + //Failed to clear index due to the index writer is closed. + Assert.fail(); + } catch (Exception ignored) { + } + //6. Do compaction and wait it completed. + TopicName segmentTopicName = NamespaceEventsSystemTopicFactory.getSystemTopicName( + TopicName.get(PROCESSOR_TOPIC).getNamespaceObject(), + EventType.TRANSACTION_BUFFER_SNAPSHOT_SEGMENTS); + TopicName indexTopicName = NamespaceEventsSystemTopicFactory.getSystemTopicName( + TopicName.get(PROCESSOR_TOPIC).getNamespaceObject(), + EventType.TRANSACTION_BUFFER_SNAPSHOT_INDEXES); + doCompaction(segmentTopicName); + doCompaction(indexTopicName); + //7. Verify the snapshot segments and index after clearing. + verifySnapshotSegmentsSize(PROCESSOR_TOPIC, 0); + verifySnapshotSegmentsIndexSize(PROCESSOR_TOPIC, 1); + } + + private void verifySnapshotSegmentsSize(String topic, int size) throws Exception { + SystemTopicClient.Reader reader = + pulsarService.getTransactionBufferSnapshotServiceFactory() + .getTxnBufferSnapshotSegmentService() + .createReader(TopicName.get(topic)).get(); + int segmentCount = 0; + while (reader.hasMoreEvents()) { + Message message = reader.readNextAsync() + .get(5, TimeUnit.SECONDS); + if (topic.equals(message.getValue().getTopicName())) { + segmentCount++; + } + } + Assert.assertEquals(segmentCount, size); + } + + private void verifySnapshotSegmentsIndexSize(String topic, int size) throws Exception { + SystemTopicClient.Reader reader = + pulsarService.getTransactionBufferSnapshotServiceFactory() + .getTxnBufferSnapshotIndexService() + .createReader(TopicName.get(topic)).get(); + int indexCount = 0; + while (reader.hasMoreEvents()) { + Message message = reader.readNextAsync() + .get(5, TimeUnit.SECONDS); + if (topic.equals(message.getValue().getTopicName())) { + indexCount++; + } + System.out.printf("message.getValue().getTopicName() :" + message.getValue().getTopicName()); + } + Assert.assertEquals(indexCount, size); + } + + private void doCompaction(TopicName topic) throws Exception { + PersistentTopic snapshotTopic = (PersistentTopic) pulsarService.getBrokerService() + .getTopic(topic.toString(), false).get().get(); + Field field = PersistentTopic.class.getDeclaredField("currentCompaction"); + field.setAccessible(true); + snapshotTopic.triggerCompaction(); + CompletableFuture compactionFuture = (CompletableFuture) field.get(snapshotTopic); + org.awaitility.Awaitility.await().untilAsserted(() -> assertTrue(compactionFuture.isDone())); + } +} 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 a2b72fc458db4..d4ddb26e014ca 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 @@ -79,6 +79,8 @@ import org.apache.pulsar.client.api.Reader; import org.apache.pulsar.client.api.ReaderBuilder; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.api.schema.GenericRecord; import org.apache.pulsar.client.api.transaction.Transaction; import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.client.impl.MessageIdImpl; @@ -91,6 +93,7 @@ import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap; import org.awaitility.Awaitility; +import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.DataProvider; @@ -530,9 +533,14 @@ public void clearTransactionBufferSnapshotTest(Boolean enableSnapshotSegment) th (AbortedTxnProcessor) abortedTxnProcessorField.get(topicTransactionBuffer); abortedTxnProcessor.takeAbortedTxnsSnapshot(topicTransactionBuffer.getMaxReadPosition()); - TopicName transactionBufferTopicName = - NamespaceEventsSystemTopicFactory.getSystemTopicName( - TopicName.get(topic).getNamespaceObject(), EventType.TRANSACTION_BUFFER_SNAPSHOT); + TopicName transactionBufferTopicName; + if (!enableSnapshotSegment) { + transactionBufferTopicName = NamespaceEventsSystemTopicFactory.getSystemTopicName( + TopicName.get(topic).getNamespaceObject(), EventType.TRANSACTION_BUFFER_SNAPSHOT); + } else { + transactionBufferTopicName = NamespaceEventsSystemTopicFactory.getSystemTopicName( + TopicName.get(topic).getNamespaceObject(), EventType.TRANSACTION_BUFFER_SNAPSHOT_INDEXES); + } PersistentTopic snapshotTopic = (PersistentTopic) getPulsarServiceList().get(0) .getBrokerService().getTopic(transactionBufferTopicName.toString(), false).get().get(); Field field = PersistentTopic.class.getDeclaredField("currentCompaction"); @@ -550,7 +558,7 @@ private void checkSnapshotCount(TopicName topicName, boolean hasSnapshot, CompletableFuture compactionFuture = (CompletableFuture) field.get(persistentTopic); Awaitility.await().untilAsserted(() -> assertTrue(compactionFuture.isDone())); - Reader reader = pulsarClient.newReader(Schema.AVRO(TransactionBufferSnapshot.class)) + Reader reader = pulsarClient.newReader(Schema.AUTO_CONSUME()) .readCompacted(true) .startMessageId(MessageId.earliest) .startMessageIdInclusive() @@ -559,7 +567,7 @@ private void checkSnapshotCount(TopicName topicName, boolean hasSnapshot, int count = 0; while (true) { - Message snapshotMsg = reader.readNext(2, TimeUnit.SECONDS); + Message snapshotMsg = reader.readNext(2, TimeUnit.SECONDS); if (snapshotMsg != null) { count++; } else { @@ -721,10 +729,10 @@ public void testTransactionBufferIndexSystemTopic() throws Exception { TransactionBufferSnapshotIndex transactionBufferSnapshotIndex = transactionBufferTransactionBufferSnapshotIndexes.getIndexList().get(1); - assertEquals(transactionBufferSnapshotIndex.getMaxReadPositionLedgerID(), 1L); - assertEquals(transactionBufferSnapshotIndex.getMaxReadPositionEntryID(), 1L); - assertEquals(transactionBufferSnapshotIndex.getPersistentPositionLedgerID(), 1L); - assertEquals(transactionBufferSnapshotIndex.getPersistentPositionEntryID(), 1L); + assertEquals(transactionBufferSnapshotIndex.getAbortedMarkLedgerID(), 1L); + assertEquals(transactionBufferSnapshotIndex.getAbortedMarkEntryID(), 1L); + assertEquals(transactionBufferSnapshotIndex.getSegmentLedgerID(), 1L); + assertEquals(transactionBufferSnapshotIndex.getSegmentEntryID(), 1L); assertEquals(transactionBufferSnapshotIndex.getSequenceID(), 1L); } @@ -765,8 +773,8 @@ public void testTransactionBufferSegmentSystemTopic() throws Exception { //build and send snapshot snapshot.setTopicName(snapshotTopic); snapshot.setSequenceId(1L); - snapshot.setMaxReadPositionLedgerId(2L); - snapshot.setMaxReadPositionEntryId(3L); + snapshot.setPersistentPositionLedgerId(2L); + snapshot.setPersistentPositionEntryId(3L); LinkedList txnIDSet = new LinkedList<>(); txnIDSet.add(new TxnIDData(1, 1)); snapshot.setAborts(txnIDSet ); @@ -818,9 +826,99 @@ public void openReadOnlyManagedLedgerFailed(ManagedLedgerException exception, Ob //verify snapshot assertEquals(snapshot.getTopicName(), snapshotTopic); assertEquals(snapshot.getSequenceId(), 2L); - assertEquals(snapshot.getMaxReadPositionLedgerId(), 2L); - assertEquals(snapshot.getMaxReadPositionEntryId(), 3L); + assertEquals(snapshot.getPersistentPositionLedgerId(), 2L); + assertEquals(snapshot.getPersistentPositionEntryId(), 3L); assertEquals(snapshot.getAborts().toArray()[0], new TxnIDData(1, 1)); } + //Verify the snapshotSegmentProcessor end to end + @Test + public void testSnapshotSegment() throws Exception { + String topic ="persistent://" + NAMESPACE1 + "/testSnapshotSegment"; + String subName = "testSnapshotSegment"; + + LinkedMap ongoingTxns = new LinkedMap<>(); + LinkedList abortedTxns = new LinkedList<>(); + // 0. Modify the configurations, enabling the segment snapshot and set the size of the snapshot segment. + int theSizeOfSegment = 10; + int theCountOfSnapshotMaxTxnCount = 3; + this.getPulsarServiceList().get(0).getConfig().setTransactionBufferSegmentedSnapshotEnabled(true); + this.getPulsarServiceList().get(0).getConfig() + .setTransactionBufferSnapshotSegmentSize(8 + topic.length() + theSizeOfSegment * 3); + this.getPulsarServiceList().get(0).getConfig() + .setTransactionBufferSnapshotMaxTransactionCount(theCountOfSnapshotMaxTxnCount); + // 1. Build producer and consumer + Producer producer = pulsarClient.newProducer(Schema.INT32) + .topic(topic) + .enableBatching(false) + .create(); + + Consumer consumer = pulsarClient.newConsumer(Schema.INT32) + .topic(topic) + .subscriptionName(subName) + .subscriptionType(SubscriptionType.Exclusive) + .subscribe(); + + // 2. Check the AbortedTxnProcessor workflow 10 times + int messageSize = theSizeOfSegment * 4; + for (int i = 0; i < 10; i++) { + MessageId maxReadMessage = null; + int abortedTxnSize = 0; + for (int j = 0; j < messageSize; j++) { + Transaction transaction = pulsarClient.newTransaction() + .withTransactionTimeout(5, TimeUnit.MINUTES).build().get(); + //Half common message and half transaction message. + if (j % 2 == 0) { + MessageId messageId = producer.newMessage(transaction).value(i * 10 + j).send(); + //And the transaction message have a half which are aborted. + if (RandomUtils.nextInt() % 2 == 0) { + transaction.abort().get(); + abortedTxns.add(messageId); + abortedTxnSize++; + } else { + ongoingTxns.put(transaction, messageId); + if (maxReadMessage == null) { + //The except number of the messages that can be read + maxReadMessage = messageId; + } + } + } else { + producer.newMessage().value(i * 10 + j).send(); + transaction.commit().get(); + } + } + // 2.1 Receive all message before the maxReadPosition to verify the correctness of the max read position. + int hasReceived = 0; + while (true) { + Message message = consumer.receive(2, TimeUnit.SECONDS); + if (message != null) { + Assert.assertTrue(message.getMessageId().compareTo(maxReadMessage) < 0); + hasReceived ++; + } else { + break; + } + } + //2.2 Commit all ongoing transaction and verify that the consumer can receive all rest message + // expect for aborted txn message. + for (Transaction ongoingTxn: ongoingTxns.keySet()) { + ongoingTxn.commit().get(); + } + ongoingTxns.clear(); + for (int k = hasReceived; k < messageSize - abortedTxnSize; k++) { + Message message = consumer.receive(2, TimeUnit.SECONDS); + assertNotNull(message); + assertFalse(abortedTxns.contains(message.getMessageId())); + } + } + // 3. After the topic unload, the consumer can receive all the messages in the 10 tests + // expect for the aborted transaction messages. + admin.topics().unload(topic); + for (int i = 0; i < messageSize * 10 - abortedTxns.size(); i++) { + Message message = consumer.receive(2, TimeUnit.SECONDS); + assertNotNull(message); + assertFalse(abortedTxns.contains(message.getMessageId())); + } + assertNull(consumer.receive(2, TimeUnit.SECONDS)); + } + }