You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/12/20 13:14:02 UTC
[pulsar] 10/15: [Transaction] Merge transactionBuffer exception into a class (#12358)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 10bbe0a0bb09f3c626cfe94eca765a3806466474
Author: Xiangying Meng <55...@users.noreply.github.com>
AuthorDate: Thu Nov 4 23:03:13 2021 +0800
[Transaction] Merge transactionBuffer exception into a class (#12358)
(cherry picked from commit 15a23030a7162ee5d5613d1b3bd4f483c1e8100f)
---
.../broker/TransactionMetadataStoreService.java | 6 +-
.../PersistentDispatcherMultipleConsumers.java | 4 +-
.../PersistentDispatcherSingleActiveConsumer.java | 4 +-
.../streamingdispatch/StreamingEntryReader.java | 4 +-
.../buffer/TransactionBufferReader.java | 5 +-
.../broker/transaction/buffer/TransactionMeta.java | 6 +-
.../NoTxnsCommittedAtLedgerException.java | 31 --------
.../exceptions/TransactionNotSealedException.java | 31 --------
.../exceptions/TransactionSealedException.java | 33 --------
.../exceptions/TransactionStatusException.java | 37 ---------
.../exceptions/UnsupportedTxnActionException.java | 34 --------
.../buffer/impl/InMemTransactionBuffer.java | 48 ++++++------
.../buffer/impl/InMemTransactionBufferReader.java | 4 +-
.../exception/TransactionException.java | 83 ++++++++++++++++++++
.../buffer/TransactionBufferException.java | 91 ++++++++++++++++++++++
.../buffer}/package-info.java | 2 +-
.../TransactionCoordinatorException.java} | 29 +++++--
.../coordinator/package-info.java} | 12 +--
.../package-info.java} | 13 +---
.../TransactionPendingAckException.java} | 25 ++++--
.../pendingack/package-info.java} | 13 +---
...ransactionPendingAckStoreProviderException.java | 32 --------
.../pendingack/impl/MLPendingAckStoreProvider.java | 5 +-
.../buffer/InMemTransactionBufferReaderTest.java | 4 +-
.../transaction/buffer/TransactionBufferTest.java | 16 ++--
25 files changed, 278 insertions(+), 294 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java
index 31a8714..607f05e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java
@@ -36,7 +36,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.pulsar.broker.namespace.NamespaceBundleOwnershipListener;
import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotReadyException;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.UnsupportedTxnActionException;
+import org.apache.pulsar.broker.transaction.exception.coordinator.TransactionCoordinatorException;
import org.apache.pulsar.broker.transaction.recover.TransactionRecoverTrackerImpl;
import org.apache.pulsar.broker.transaction.timeout.TransactionTimeoutTrackerFactoryImpl;
import org.apache.pulsar.client.api.PulsarClientException.BrokerPersistenceException;
@@ -331,8 +331,8 @@ public class TransactionMetadataStoreService {
newStatus = ABORTING;
break;
default:
- UnsupportedTxnActionException exception =
- new UnsupportedTxnActionException(txnID, txnAction);
+ TransactionCoordinatorException.UnsupportedTxnActionException exception =
+ new TransactionCoordinatorException.UnsupportedTxnActionException(txnID, txnAction);
LOG.error(exception.getMessage());
completableFuture.completeExceptionally(exception);
return completableFuture;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
index 80e3d60..0d06005 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -55,7 +55,7 @@ import org.apache.pulsar.broker.service.SendMessageInfo;
import org.apache.pulsar.broker.service.StickyKeyConsumerSelector;
import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotSealedException;
+import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException;
import org.apache.pulsar.client.impl.Backoff;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
import org.apache.pulsar.common.api.proto.MessageMetadata;
@@ -610,7 +610,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
// Notify the consumer only if all the messages were already acknowledged
consumerList.forEach(Consumer::reachedEndOfTopic);
}
- } else if (exception.getCause() instanceof TransactionNotSealedException) {
+ } else if (exception.getCause() instanceof TransactionBufferException.TransactionNotSealedException) {
waitTimeMillis = 1;
if (log.isDebugEnabled()) {
log.debug("[{}] Error reading transaction entries : {}, Read Type {} - Retrying to read in {} seconds",
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
index 653c1c1..848cbb4 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -45,7 +45,7 @@ import org.apache.pulsar.broker.service.RedeliveryTrackerDisabled;
import org.apache.pulsar.broker.service.SendMessageInfo;
import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotSealedException;
+import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException;
import org.apache.pulsar.client.impl.Backoff;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
import org.apache.pulsar.common.policies.data.DispatchRate;
@@ -469,7 +469,7 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
// Notify the consumer only if all the messages were already acknowledged
consumers.forEach(Consumer::reachedEndOfTopic);
}
- } else if (exception.getCause() instanceof TransactionNotSealedException) {
+ } else if (exception.getCause() instanceof TransactionBufferException.TransactionNotSealedException) {
waitTimeMillis = 1;
if (log.isDebugEnabled()) {
log.debug("[{}] Error reading transaction entries : {}, - Retrying to read in {} seconds", name,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/streamingdispatch/StreamingEntryReader.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/streamingdispatch/StreamingEntryReader.java
index 12f5600..85083da 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/streamingdispatch/StreamingEntryReader.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/streamingdispatch/StreamingEntryReader.java
@@ -37,7 +37,7 @@ import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.bookkeeper.mledger.util.SafeRun;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotSealedException;
+import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException;
import org.apache.pulsar.client.impl.Backoff;
/**
@@ -197,7 +197,7 @@ public class StreamingEntryReader implements AsyncCallbacks.ReadEntryCallback, W
PositionImpl readPosition = pendingReadEntryRequest.position;
pendingReadEntryRequest.retry++;
long waitTimeMillis = readFailureBackoff.next();
- if (exception.getCause() instanceof TransactionNotSealedException) {
+ if (exception.getCause() instanceof TransactionBufferException.TransactionNotSealedException) {
waitTimeMillis = 1;
if (log.isDebugEnabled()) {
log.debug("[{}] Error reading transaction entries : {}, - Retrying to read in {} seconds",
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferReader.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferReader.java
index 4547de0..54dafd5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferReader.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferReader.java
@@ -21,7 +21,8 @@ package org.apache.pulsar.broker.transaction.buffer;
import com.google.common.annotations.Beta;
import java.util.List;
import java.util.concurrent.CompletableFuture;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.EndOfTransactionException;
+import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException;
+
/**
* A reader to read entries of a given transaction from transaction buffer.
@@ -38,7 +39,7 @@ public interface TransactionBufferReader extends AutoCloseable {
*
* @param numEntries the number of entries to read from transaction buffer.
* @return a future represents the result of the read operations.
- * @throws EndOfTransactionException if reaching end of the transaction and no
+ * @throws TransactionBufferException.EndOfTransactionException if reaching end of the transaction and no
* more entries to return.
*/
CompletableFuture<List<TransactionEntry>> readNext(int numEntries);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionMeta.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionMeta.java
index 9afc05e..347ab74 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionMeta.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionMeta.java
@@ -22,7 +22,7 @@ import com.google.common.annotations.Beta;
import java.util.SortedMap;
import java.util.concurrent.CompletableFuture;
import org.apache.bookkeeper.mledger.Position;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionStatusException;
+import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.transaction.coordinator.proto.TxnStatus;
@@ -57,9 +57,9 @@ public interface TransactionMeta {
* Return messages number in one transaction.
*
* @return the number of transaction messages
- * @throws TransactionStatusException
+ * @throws TransactionBufferException.TransactionStatusException
*/
- int numMessageInTxn() throws TransactionStatusException;
+ int numMessageInTxn() throws TransactionBufferException.TransactionStatusException;
/**
* Return the committed ledger id at data ledger.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/NoTxnsCommittedAtLedgerException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/NoTxnsCommittedAtLedgerException.java
deleted file mode 100644
index b20b29a..0000000
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/NoTxnsCommittedAtLedgerException.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/**
- * 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.exceptions;
-
-/**
- * Exception is thrown when no transactions found committed at a given ledger.
- */
-public class NoTxnsCommittedAtLedgerException extends TransactionBufferException {
-
- private static final long serialVersionUID = 0L;
-
- public NoTxnsCommittedAtLedgerException(String message) {
- super(message);
- }
-}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionNotSealedException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionNotSealedException.java
deleted file mode 100644
index c732c88..0000000
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionNotSealedException.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/**
- * 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.exceptions;
-
-/**
- * Exception is thrown when opening a reader on a transaction that is not sealed yet.
- */
-public class TransactionNotSealedException extends TransactionBufferException {
-
- private static final long serialVersionUID = 0L;
-
- public TransactionNotSealedException(String message) {
- super(message);
- }
-}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionSealedException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionSealedException.java
deleted file mode 100644
index 13fdec5..0000000
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionSealedException.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/**
- * 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.exceptions;
-
-/**
- * Exception thrown if a transaction is already sealed.
- *
- * <p>If a transaction is sealed, no more entries should be appended to this transaction.
- */
-public class TransactionSealedException extends TransactionBufferException {
-
- private static final long serialVersionUID = 5366602873819540477L;
-
- public TransactionSealedException(String message) {
- super(message);
- }
-}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionStatusException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionStatusException.java
deleted file mode 100644
index 008f27f..0000000
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionStatusException.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/**
- * 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.exceptions;
-
-import org.apache.pulsar.client.api.transaction.TxnID;
-import org.apache.pulsar.transaction.coordinator.proto.TxnStatus;
-
-/**
- * Exceptions are thrown when operations are applied to a transaction which is not in expected txn status.
- */
-public class TransactionStatusException extends TransactionBufferException {
-
- private static final long serialVersionUID = 0L;
-
- public TransactionStatusException(TxnID txnId,
- TxnStatus expectedStatus,
- TxnStatus actualStatus) {
- super("Transaction `" + txnId + "` is not in an expected status `" + expectedStatus
- + "`, but is in status `" + actualStatus + "`");
- }
-}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/UnsupportedTxnActionException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/UnsupportedTxnActionException.java
deleted file mode 100644
index e4a8341..0000000
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/UnsupportedTxnActionException.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/**
- * 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.exceptions;
-
-import org.apache.pulsar.client.api.transaction.TxnID;
-import org.apache.pulsar.common.api.proto.TxnAction;
-
-/**
- * Exceptions are thrown when txnAction is unsupported.
- */
-public class UnsupportedTxnActionException extends TransactionBufferException {
-
- private static final long serialVersionUID = 0L;
-
- public UnsupportedTxnActionException(TxnID txnId, int txnAction) {
- super("Transaction `" + txnId + "` receive unsupported txnAction " + TxnAction.valueOf(txnAction));
- }
-}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java
index 2c32d0b..f66b263 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java
@@ -36,10 +36,7 @@ import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.transaction.buffer.TransactionBuffer;
import org.apache.pulsar.broker.transaction.buffer.TransactionBufferReader;
import org.apache.pulsar.broker.transaction.buffer.TransactionMeta;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotFoundException;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotSealedException;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionSealedException;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionStatusException;
+import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.common.policies.data.TransactionBufferStats;
import org.apache.pulsar.common.policies.data.TransactionInBufferStats;
@@ -86,7 +83,7 @@ class InMemTransactionBuffer implements TransactionBuffer {
}
@Override
- public int numMessageInTxn() throws TransactionStatusException {
+ public int numMessageInTxn() throws TransactionBufferException.TransactionStatusException {
return -1;
}
@@ -125,7 +122,7 @@ class InMemTransactionBuffer implements TransactionBuffer {
public CompletableFuture<TransactionMeta> commitTxn(long committedAtLedgerId, long committedAtEntryId) {
try {
return CompletableFuture.completedFuture(commitAt(committedAtLedgerId, committedAtEntryId));
- } catch (TransactionStatusException e) {
+ } catch (TransactionBufferException.TransactionStatusException e) {
return FutureUtil.failedFuture(e);
}
}
@@ -134,23 +131,23 @@ class InMemTransactionBuffer implements TransactionBuffer {
public CompletableFuture<TransactionMeta> abortTxn() {
try {
return CompletableFuture.completedFuture(abort());
- } catch (TransactionStatusException e) {
+ } catch (TransactionBufferException.TransactionStatusException e) {
return FutureUtil.failedFuture(e);
}
}
- synchronized TxnBuffer abort() throws TransactionStatusException {
+ synchronized TxnBuffer abort() throws TransactionBufferException.TransactionStatusException {
if (TxnStatus.OPEN != status) {
- throw new TransactionStatusException(txnid, TxnStatus.OPEN, status);
+ throw new TransactionBufferException.TransactionStatusException(txnid, TxnStatus.OPEN, status);
}
this.status = TxnStatus.ABORTED;
return this;
}
synchronized TxnBuffer commitAt(long committedAtLedgerId, long committedAtEntryId)
- throws TransactionStatusException {
+ throws TransactionBufferException.TransactionStatusException {
if (TxnStatus.OPEN != status) {
- throw new TransactionStatusException(txnid, TxnStatus.OPEN, status);
+ throw new TransactionBufferException.TransactionStatusException(txnid, TxnStatus.OPEN, status);
}
this.committedAtLedgerId = committedAtLedgerId;
@@ -170,11 +167,13 @@ class InMemTransactionBuffer implements TransactionBuffer {
}
}
- public void appendEntry(long sequenceId, ByteBuf entry) throws TransactionSealedException {
+ public void appendEntry(long sequenceId, ByteBuf entry) throws
+ TransactionBufferException.TransactionSealedException {
synchronized (this) {
if (TxnStatus.OPEN != status) {
// the transaction is not open anymore, reject the append operations
- throw new TransactionSealedException("Transaction `" + txnid + "` is already sealed");
+ throw new TransactionBufferException
+ .TransactionSealedException("Transaction `" + txnid + "` is already sealed");
}
}
@@ -183,11 +182,13 @@ class InMemTransactionBuffer implements TransactionBuffer {
}
}
- public TransactionBufferReader newReader(long sequenceId) throws TransactionNotSealedException {
+ public TransactionBufferReader newReader(long sequenceId) throws
+ TransactionBufferException.TransactionNotSealedException {
synchronized (this) {
if (TxnStatus.COMMITTED != status) {
// the transaction is not committed yet, hence the buffer is not sealed
- throw new TransactionNotSealedException("Transaction `" + txnid + "` is not sealed yet");
+ throw new TransactionBufferException
+ .TransactionNotSealedException("Transaction `" + txnid + "` is not sealed yet");
}
}
@@ -220,17 +221,17 @@ class InMemTransactionBuffer implements TransactionBuffer {
CompletableFuture<TransactionMeta> getFuture = new CompletableFuture<>();
try {
getFuture.complete(getTxnBufferOrThrowNotFoundException(txnID));
- } catch (TransactionNotFoundException e) {
+ } catch (TransactionBufferException.TransactionNotFoundException e) {
getFuture.completeExceptionally(e);
}
return getFuture;
}
private TxnBuffer getTxnBufferOrThrowNotFoundException(TxnID txnID)
- throws TransactionNotFoundException {
+ throws TransactionBufferException.TransactionNotFoundException {
TxnBuffer buffer = buffers.get(txnID);
if (null == buffer) {
- throw new TransactionNotFoundException(
+ throw new TransactionBufferException.TransactionNotFoundException(
"Transaction `" + txnID + "` doesn't exist in the transaction buffer");
}
return buffer;
@@ -262,7 +263,7 @@ class InMemTransactionBuffer implements TransactionBuffer {
try {
txnBuffer.appendEntry(sequenceId, buffer);
appendFuture.complete(null);
- } catch (TransactionSealedException e) {
+ } catch (TransactionBufferException.TransactionSealedException e) {
appendFuture.completeExceptionally(e);
}
return appendFuture;
@@ -276,7 +277,8 @@ class InMemTransactionBuffer implements TransactionBuffer {
TxnBuffer txnBuffer = getTxnBufferOrThrowNotFoundException(txnID);
TransactionBufferReader reader = txnBuffer.newReader(startSequenceId);
openFuture.complete(reader);
- } catch (TransactionNotFoundException | TransactionNotSealedException e) {
+ } catch (TransactionBufferException.TransactionNotFoundException
+ | TransactionBufferException.TransactionNotSealedException e) {
openFuture.completeExceptionally(e);
}
return openFuture;
@@ -295,7 +297,8 @@ class InMemTransactionBuffer implements TransactionBuffer {
addTxnToTxnIdex(txnID, committedAtLedgerId);
}
commitFuture.complete(null);
- } catch (TransactionNotFoundException | TransactionStatusException e) {
+ } catch (TransactionBufferException.TransactionNotFoundException
+ | TransactionBufferException.TransactionStatusException e) {
commitFuture.completeExceptionally(e);
}
return commitFuture;
@@ -318,7 +321,8 @@ class InMemTransactionBuffer implements TransactionBuffer {
txnBuffer.abort();
buffers.remove(txnID, txnBuffer);
abortFuture.complete(null);
- } catch (TransactionNotFoundException | TransactionStatusException e) {
+ } catch (TransactionBufferException.TransactionNotFoundException
+ | TransactionBufferException.TransactionStatusException e) {
abortFuture.completeExceptionally(e);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBufferReader.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBufferReader.java
index 91b095e..d81ce02 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBufferReader.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBufferReader.java
@@ -27,7 +27,7 @@ import java.util.concurrent.CompletableFuture;
import org.apache.bookkeeper.mledger.impl.EntryImpl;
import org.apache.pulsar.broker.transaction.buffer.TransactionBufferReader;
import org.apache.pulsar.broker.transaction.buffer.TransactionEntry;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.EndOfTransactionException;
+import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException;
import org.apache.pulsar.client.api.transaction.TxnID;
/**
@@ -78,7 +78,7 @@ public class InMemTransactionBufferReader implements TransactionBufferReader {
}
if (txnEntries.isEmpty()) {
- readFuture.completeExceptionally(new EndOfTransactionException(
+ readFuture.completeExceptionally(new TransactionBufferException.EndOfTransactionException(
"No more entries found in transaction `" + txnId + "`"
));
} else {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/TransactionException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/TransactionException.java
new file mode 100644
index 0000000..f81fc8a
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/TransactionException.java
@@ -0,0 +1,83 @@
+/**
+ * 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.exception;
+
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.transaction.coordinator.proto.TxnStatus;
+
+/**
+ * The base exception class for the errors thrown from Transaction.
+ */
+public abstract class TransactionException extends Exception {
+
+ private static final long serialVersionUID = 0L;
+
+ public TransactionException(String message) {
+ super(message);
+ }
+
+ public TransactionException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public TransactionException(Throwable cause) {
+ super(cause);
+ }
+
+ /**
+ * Exception is thrown when opening a reader on a transaction that is not sealed yet.
+ */
+ public static class TransactionNotSealedException extends TransactionException {
+
+ private static final long serialVersionUID = 0L;
+
+ public TransactionNotSealedException(String message) {
+ super(message);
+ }
+ }
+
+ /**
+ * Exception thrown if a transaction is already sealed.
+ *
+ * <p>If a transaction is sealed, no more entries should be appended to this transaction.
+ */
+ public static class TransactionSealedException extends TransactionException {
+
+ private static final long serialVersionUID = 5366602873819540477L;
+
+ public TransactionSealedException(String message) {
+ super(message);
+ }
+ }
+
+ /**
+ * Exceptions are thrown when operations are applied to a transaction which is not in expected txn status.
+ */
+ public static class TransactionStatusException extends TransactionException {
+
+ private static final long serialVersionUID = 0L;
+
+ public TransactionStatusException(TxnID txnId,
+ TxnStatus expectedStatus,
+ TxnStatus actualStatus) {
+ super("Transaction `q" + txnId + "` is not in an expected status `" + expectedStatus
+ + "`, but is in status `" + actualStatus + "`");
+ }
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/buffer/TransactionBufferException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/buffer/TransactionBufferException.java
new file mode 100644
index 0000000..ab301c9
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/buffer/TransactionBufferException.java
@@ -0,0 +1,91 @@
+/**
+ * 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.exception.buffer;
+
+import org.apache.pulsar.broker.transaction.exception.TransactionException;
+
+/**
+ * The base exception class for the errors thrown from Transaction Buffer.
+ */
+public abstract class TransactionBufferException extends TransactionException {
+
+ private static final long serialVersionUID = 0L;
+
+ public TransactionBufferException(String message) {
+ super(message);
+ }
+
+ public TransactionBufferException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public TransactionBufferException(Throwable cause) {
+ super(cause);
+ }
+
+
+ /**
+ * Exception thrown when reaching end of a transaction.
+ */
+ public static class EndOfTransactionException extends TransactionBufferException {
+
+ private static final long serialVersionUID = 0L;
+
+ public EndOfTransactionException(String message) {
+ super(message);
+ }
+ }
+
+ /**
+ * Exception is thrown when no transactions found committed at a given ledger.
+ */
+ public class NoTxnsCommittedAtLedgerException extends TransactionBufferException {
+
+ private static final long serialVersionUID = 0L;
+
+ public NoTxnsCommittedAtLedgerException(String message) {
+ super(message);
+ }
+ }
+
+ /**
+ * Transaction buffer provider exception.
+ */
+ public class TransactionBufferProviderException extends TransactionBufferException {
+
+ public TransactionBufferProviderException(String message) {
+ super(message);
+ }
+
+ }
+
+ /**
+ * Exception is thrown when the transaction is not found in the transaction buffer.
+ */
+ public static class TransactionNotFoundException extends TransactionBufferException {
+
+ private static final long serialVersionUID = 0L;
+
+ public TransactionNotFoundException(String message) {
+ super(message);
+ }
+ }
+
+
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/package-info.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/buffer/package-info.java
similarity index 93%
rename from pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/package-info.java
rename to pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/buffer/package-info.java
index 2aee740..60bb5ee2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/package-info.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/buffer/package-info.java
@@ -19,4 +19,4 @@
/**
* Exceptions thrown when encountering errors in transaction buffer.
*/
-package org.apache.pulsar.broker.transaction.buffer.exceptions;
+package org.apache.pulsar.broker.transaction.exception.buffer;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionBufferException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/coordinator/TransactionCoordinatorException.java
similarity index 51%
copy from pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionBufferException.java
copy to pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/coordinator/TransactionCoordinatorException.java
index 3ecbf2b..8da04ac 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionBufferException.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/coordinator/TransactionCoordinatorException.java
@@ -16,24 +16,41 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.broker.transaction.buffer.exceptions;
+package org.apache.pulsar.broker.transaction.exception.coordinator;
+
+import org.apache.pulsar.broker.transaction.exception.TransactionException;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.api.proto.TxnAction;
/**
- * The base exception class for the errors thrown from Transaction Buffer.
+ * The base exception class for the errors thrown from Transaction Coordinator.
*/
-public abstract class TransactionBufferException extends Exception {
+public abstract class TransactionCoordinatorException extends TransactionException {
private static final long serialVersionUID = 0L;
- public TransactionBufferException(String message) {
+ public TransactionCoordinatorException(String message) {
super(message);
}
- public TransactionBufferException(String message, Throwable cause) {
+ public TransactionCoordinatorException(String message, Throwable cause) {
super(message, cause);
}
- public TransactionBufferException(Throwable cause) {
+ public TransactionCoordinatorException(Throwable cause) {
super(cause);
}
+
+
+ /**
+ * Exceptions are thrown when txnAction is unsupported.
+ */
+ public static class UnsupportedTxnActionException extends TransactionCoordinatorException {
+
+ private static final long serialVersionUID = 0L;
+
+ public UnsupportedTxnActionException(TxnID txnId, int txnAction) {
+ super("Transaction `" + txnId + "` receive unsupported txnAction " + TxnAction.valueOf(txnAction));
+ }
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionBufferProviderException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/coordinator/package-info.java
similarity index 73%
rename from pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionBufferProviderException.java
rename to pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/coordinator/package-info.java
index e7a7a62..ceaff6a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionBufferProviderException.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/coordinator/package-info.java
@@ -16,15 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.broker.transaction.buffer.exceptions;
-
/**
- * Transaction buffer provider exception.
+ * Exceptions thrown when encountering errors in transaction buffer.
*/
-public class TransactionBufferProviderException extends TransactionBufferException {
-
- public TransactionBufferProviderException(String message) {
- super(message);
- }
-
-}
+package org.apache.pulsar.broker.transaction.exception.coordinator;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/EndOfTransactionException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/package-info.java
similarity index 70%
rename from pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/EndOfTransactionException.java
rename to pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/package-info.java
index 57fce9a..222d871 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/EndOfTransactionException.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/package-info.java
@@ -16,16 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.broker.transaction.buffer.exceptions;
-
/**
- * Exception thrown when reaching end of a transaction.
+ * Exceptions thrown when encountering errors in transaction buffer.
*/
-public class EndOfTransactionException extends TransactionBufferException {
-
- private static final long serialVersionUID = 0L;
-
- public EndOfTransactionException(String message) {
- super(message);
- }
-}
+package org.apache.pulsar.broker.transaction.exception;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionBufferException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/pendingack/TransactionPendingAckException.java
similarity index 58%
rename from pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionBufferException.java
rename to pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/pendingack/TransactionPendingAckException.java
index 3ecbf2b..74999bf 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionBufferException.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/pendingack/TransactionPendingAckException.java
@@ -16,24 +16,37 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.broker.transaction.buffer.exceptions;
+package org.apache.pulsar.broker.transaction.exception.pendingack;
+
+import org.apache.pulsar.broker.transaction.exception.TransactionException;
/**
- * The base exception class for the errors thrown from Transaction Buffer.
+ * The base exception class for the errors thrown from Transaction Pending ACk.
*/
-public abstract class TransactionBufferException extends Exception {
+public abstract class TransactionPendingAckException extends TransactionException {
private static final long serialVersionUID = 0L;
- public TransactionBufferException(String message) {
+ public TransactionPendingAckException(String message) {
super(message);
}
- public TransactionBufferException(String message, Throwable cause) {
+ public TransactionPendingAckException(String message, Throwable cause) {
super(message, cause);
}
- public TransactionBufferException(Throwable cause) {
+ public TransactionPendingAckException(Throwable cause) {
super(cause);
}
+
+ /**
+ * Transaction pending ack store provider exception.
+ */
+ public static class TransactionPendingAckStoreProviderException extends TransactionPendingAckException {
+
+ public TransactionPendingAckStoreProviderException(String message) {
+ super(message);
+ }
+
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionNotFoundException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/pendingack/package-info.java
similarity index 68%
rename from pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionNotFoundException.java
rename to pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/pendingack/package-info.java
index 0f1dc46..0d0df63 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionNotFoundException.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/exception/pendingack/package-info.java
@@ -16,16 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.broker.transaction.buffer.exceptions;
-
/**
- * Exception is thrown when the transaction is not found in the transaction buffer.
+ * Exceptions thrown when encountering errors in transaction buffer.
*/
-public class TransactionNotFoundException extends TransactionBufferException {
-
- private static final long serialVersionUID = 0L;
-
- public TransactionNotFoundException(String message) {
- super(message);
- }
-}
+package org.apache.pulsar.broker.transaction.exception.pendingack;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/exceptions/TransactionPendingAckStoreProviderException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/exceptions/TransactionPendingAckStoreProviderException.java
deleted file mode 100644
index 0f41e2c..0000000
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/exceptions/TransactionPendingAckStoreProviderException.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/**
- * 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.pendingack.exceptions;
-
-import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionBufferException;
-
-/**
- * Transaction pending ack store provider exception.
- */
-public class TransactionPendingAckStoreProviderException extends TransactionBufferException {
-
- public TransactionPendingAckStoreProviderException(String message) {
- super(message);
- }
-
-}
\ No newline at end of file
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreProvider.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreProvider.java
index 548dd8a..5417cae 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreProvider.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreProvider.java
@@ -26,9 +26,9 @@ import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.transaction.exception.pendingack.TransactionPendingAckException;
import org.apache.pulsar.broker.transaction.pendingack.PendingAckStore;
import org.apache.pulsar.broker.transaction.pendingack.TransactionPendingAckStoreProvider;
-import org.apache.pulsar.broker.transaction.pendingack.exceptions.TransactionPendingAckStoreProviderException;
import org.apache.pulsar.common.api.proto.CommandSubscribe.InitialPosition;
import org.apache.pulsar.common.naming.TopicName;
@@ -45,7 +45,8 @@ public class MLPendingAckStoreProvider implements TransactionPendingAckStoreProv
if (subscription == null) {
pendingAckStoreFuture.completeExceptionally(
- new TransactionPendingAckStoreProviderException("The subscription is null."));
+ new TransactionPendingAckException
+ .TransactionPendingAckStoreProviderException("The subscription is null."));
return pendingAckStoreFuture;
}
PersistentTopic originPersistentTopic = (PersistentTopic) subscription.getTopic();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/InMemTransactionBufferReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/InMemTransactionBufferReaderTest.java
index 513a78d..d4234fb 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/InMemTransactionBufferReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/InMemTransactionBufferReaderTest.java
@@ -33,7 +33,7 @@ import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.EndOfTransactionException;
+import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException;
import org.apache.pulsar.broker.transaction.buffer.impl.InMemTransactionBufferReader;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.testng.annotations.Test;
@@ -116,7 +116,7 @@ public class InMemTransactionBufferReaderTest {
reader.readNext(1).get();
fail("should fail to read entries if there is no more in the transaction buffer");
} catch (ExecutionException ee) {
- assertTrue(ee.getCause() instanceof EndOfTransactionException);
+ assertTrue(ee.getCause() instanceof TransactionBufferException.EndOfTransactionException);
}
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferTest.java
index 2238490..c10444a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferTest.java
@@ -33,11 +33,9 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException;
import org.apache.pulsar.broker.transaction.buffer.impl.InMemTransactionBufferProvider;
import org.apache.pulsar.client.api.transaction.TxnID;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotFoundException;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotSealedException;
-import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionStatusException;
import org.apache.pulsar.transaction.coordinator.proto.TxnStatus;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -86,7 +84,7 @@ public class TransactionBufferTest {
buffer.openTransactionBufferReader(txnId, 0L).get();
fail("Should fail to open reader if a transaction doesn't exist");
} catch (ExecutionException ee) {
- assertTrue(ee.getCause() instanceof TransactionNotFoundException);
+ assertTrue(ee.getCause() instanceof TransactionBufferException.TransactionNotFoundException);
}
}
@@ -102,7 +100,7 @@ public class TransactionBufferTest {
buffer.openTransactionBufferReader(txnId, 0L).get();
fail("Should fail to open a reader on an OPEN transaction");
} catch (ExecutionException e) {
- assertTrue(e.getCause() instanceof TransactionNotSealedException);
+ assertTrue(e.getCause() instanceof TransactionBufferException.TransactionNotSealedException);
}
}
@@ -136,7 +134,7 @@ public class TransactionBufferTest {
buffer.commitTxn(txnId, Long.MIN_VALUE).get();
fail("Should fail to commit a transaction if it doesn't exist");
} catch (ExecutionException ee) {
- assertTrue(ee.getCause() instanceof TransactionNotFoundException);
+ assertTrue(ee.getCause() instanceof TransactionBufferException.TransactionNotFoundException);
}
}
@@ -160,7 +158,7 @@ public class TransactionBufferTest {
buffer.abortTxn(txnId, Long.MIN_VALUE).get();
fail("Should fail to abort a transaction if it doesn't exist");
} catch (ExecutionException ee) {
- assertTrue(ee.getCause() instanceof TransactionNotFoundException);
+ assertTrue(ee.getCause() instanceof TransactionBufferException.TransactionNotFoundException);
}
}
@@ -181,7 +179,7 @@ public class TransactionBufferTest {
buffer.abortTxn(txnId, Long.MIN_VALUE).get();
fail("Should fail to abort a committed transaction");
} catch (ExecutionException e) {
- assertTrue(e.getCause() instanceof TransactionStatusException);
+ assertTrue(e.getCause() instanceof TransactionBufferException.TransactionStatusException);
}
txnMeta = buffer.getTransactionMeta(txnId).get();
assertEquals(txnId, txnMeta.id());
@@ -277,7 +275,7 @@ public class TransactionBufferTest {
buffer.getTransactionMeta(txnID).get();
fail("Should fail to get transaction metadata if it doesn't exist");
} catch (ExecutionException ee) {
- assertTrue(ee.getCause() instanceof TransactionNotFoundException);
+ assertTrue(ee.getCause() instanceof TransactionBufferException.TransactionNotFoundException);
}
}
}