[Transaction][Buffer] move the transaction buffer code to the broker …

…module (apache#4919)


*Motivation*

Currently, all transaction buffer code in the transaction module. A transaction buffer inherits `PersistentTopic`, it depends on the pulsar-broker module. When we need to create a transaction buffer at the pulsar-broker, the pulsar-broker depends on the transaction module. So it will cause cyclic reference.

*Modifications*

- move the code under transaction buffer module to the broker module

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/TransactionBuffer.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionBuffer.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer;
+package org.apache.pulsar.broker.transaction.buffer;
 
 import com.google.common.annotations.Beta;
 import io.netty.buffer.ByteBuf;
@@ -52,7 +52,7 @@ public interface TransactionBuffer {
      *
      * @param txnID the transaction id
      * @return a future represents the result of the operation
-     * @throws org.apache.pulsar.transaction.buffer.exceptions.TransactionNotFoundException if the transaction
+     * @throws org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotFoundException if the transaction
      *         is not in the buffer.
      */
     CompletableFuture<TransactionMeta> getTransactionMeta(TxnID txnID);
@@ -66,7 +66,7 @@ public interface TransactionBuffer {
      * @param sequenceId the sequence id of the entry in this transaction buffer.
      * @param buffer the entry buffer
      * @return a future represents the result of the operation.
-     * @throws org.apache.pulsar.transaction.buffer.exceptions.TransactionSealedException if the transaction
+     * @throws org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionSealedException if the transaction
      *         has been sealed.
      */
     CompletableFuture<Void> appendBufferToTxn(TxnID txnId, long sequenceId, ByteBuf buffer);
@@ -78,7 +78,7 @@ public interface TransactionBuffer {
      * @param txnID transaction id
      * @param startSequenceId the sequence id to start read
      * @return a future represents the result of open operation.
-     * @throws org.apache.pulsar.transaction.buffer.exceptions.TransactionNotFoundException if the transaction
+     * @throws org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotFoundException if the transaction
      *         is not in the buffer.
      */
     CompletableFuture<TransactionBufferReader> openTransactionBufferReader(TxnID txnID, long startSequenceId);
@@ -92,7 +92,7 @@ public interface TransactionBuffer {
      * @param committedAtLedgerId the data ledger id where the commit marker of the transaction was appended to.
      * @param committedAtEntryId the data ledger id where the commit marker of the transaction was appended to.
      * @return a future represents the result of commit operation.
-     * @throws org.apache.pulsar.transaction.buffer.exceptions.TransactionNotFoundException if the transaction
+     * @throws org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotFoundException if the transaction
      *         is not in the buffer.
      */
     CompletableFuture<Void> commitTxn(TxnID txnID, long committedAtLedgerId, long committedAtEntryId);
@@ -103,7 +103,7 @@ public interface TransactionBuffer {
      *
      * @param txnID the transaction id
      * @return a future represents the result of abort operation.
-     * @throws org.apache.pulsar.transaction.buffer.exceptions.TransactionNotFoundException if the transaction
+     * @throws org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotFoundException if the transaction
      *         is not in the buffer.
      */
     CompletableFuture<Void> abortTxn(TxnID txnID);

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/TransactionBufferProvider.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferProvider.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer;
+package org.apache.pulsar.broker.transaction.buffer;
 
 import static com.google.common.base.Preconditions.checkArgument;
 

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/TransactionBufferReader.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferReader.java

@@ -16,12 +16,12 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer;
+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.transaction.buffer.exceptions.EndOfTransactionException;
+import org.apache.pulsar.broker.transaction.buffer.exceptions.EndOfTransactionException;
 
 /**
  * A reader to read entries of a given transaction from transaction buffer.

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/TransactionCursor.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionCursor.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer;
+package org.apache.pulsar.broker.transaction.buffer;
 
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/TransactionEntry.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionEntry.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer;
+package org.apache.pulsar.broker.transaction.buffer;
 
 import com.google.common.annotations.Beta;
 import io.netty.buffer.ByteBuf;

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/TransactionMeta.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionMeta.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer;
+package org.apache.pulsar.broker.transaction.buffer;
 
 import com.google.common.annotations.Beta;
 import java.util.SortedMap;

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/exceptions/EndOfTransactionException.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/EndOfTransactionException.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.exceptions;
+package org.apache.pulsar.broker.transaction.buffer.exceptions;
 
 /**
  * Exception thrown when reaching end of a transaction.

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/exceptions/NoTxnsCommittedAtLedgerException.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/NoTxnsCommittedAtLedgerException.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.exceptions;
+package org.apache.pulsar.broker.transaction.buffer.exceptions;
 
 /**
  * Exception is thrown when no transactions found committed at a given ledger.

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/exceptions/TransactionBufferException.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionBufferException.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.exceptions;
+package org.apache.pulsar.broker.transaction.buffer.exceptions;
 
 /**
  * The base exception class for the errors thrown from Transaction Buffer.

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/exceptions/TransactionNotFoundException.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionNotFoundException.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.exceptions;
+package org.apache.pulsar.broker.transaction.buffer.exceptions;
 
 /**
  * Exception is thrown when the transaction is not found in the transaction buffer.

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/exceptions/TransactionNotSealedException.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionNotSealedException.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.exceptions;
+package org.apache.pulsar.broker.transaction.buffer.exceptions;
 
 /**
  * Exception is thrown when opening a reader on a transaction that is not sealed yet.

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/exceptions/TransactionSealedException.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/TransactionSealedException.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.exceptions;
+package org.apache.pulsar.broker.transaction.buffer.exceptions;
 
 /**
  * Exception thrown if a transaction is already sealed.

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/exceptions/UnexpectedTxnStatusException.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/UnexpectedTxnStatusException.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.exceptions;
+package org.apache.pulsar.broker.transaction.buffer.exceptions;
 
 import org.apache.pulsar.transaction.impl.common.TxnID;
 import org.apache.pulsar.transaction.impl.common.TxnStatus;

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/exceptions/package-info.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/exceptions/package-info.java

@@ -19,4 +19,4 @@
 /**
  * Exceptions thrown when encountering errors in transaction buffer.
  */
-package org.apache.pulsar.transaction.buffer.exceptions;
+package org.apache.pulsar.broker.transaction.buffer.exceptions;

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/impl/InMemTransactionBuffer.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.impl;
+package org.apache.pulsar.broker.transaction.buffer.impl;
 
 import io.netty.buffer.ByteBuf;
 import java.util.ArrayList;
@@ -32,13 +32,13 @@
 import java.util.concurrent.ConcurrentMap;
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.pulsar.common.util.FutureUtil;
-import org.apache.pulsar.transaction.buffer.TransactionBuffer;
-import org.apache.pulsar.transaction.buffer.TransactionBufferReader;
-import org.apache.pulsar.transaction.buffer.TransactionMeta;
-import org.apache.pulsar.transaction.buffer.exceptions.TransactionNotFoundException;
-import org.apache.pulsar.transaction.buffer.exceptions.TransactionNotSealedException;
-import org.apache.pulsar.transaction.buffer.exceptions.TransactionSealedException;
-import org.apache.pulsar.transaction.buffer.exceptions.UnexpectedTxnStatusException;
+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.UnexpectedTxnStatusException;
 import org.apache.pulsar.transaction.impl.common.TxnID;
 import org.apache.pulsar.transaction.impl.common.TxnStatus;
 

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/impl/InMemTransactionBufferProvider.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBufferProvider.java

@@ -16,11 +16,11 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.impl;
+package org.apache.pulsar.broker.transaction.buffer.impl;
 
 import java.util.concurrent.CompletableFuture;
-import org.apache.pulsar.transaction.buffer.TransactionBuffer;
-import org.apache.pulsar.transaction.buffer.TransactionBufferProvider;
+import org.apache.pulsar.broker.transaction.buffer.TransactionBuffer;
+import org.apache.pulsar.broker.transaction.buffer.TransactionBufferProvider;
 
 /**
  * A provider that provides in-memory implementations of {@link TransactionBuffer}.

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/impl/InMemTransactionBufferReader.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBufferReader.java

@@ -16,23 +16,23 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.impl;
+package org.apache.pulsar.broker.transaction.buffer.impl;
 
 import io.netty.buffer.ByteBuf;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map.Entry;
 import java.util.concurrent.CompletableFuture;
-import org.apache.pulsar.transaction.buffer.TransactionBufferReader;
-import org.apache.pulsar.transaction.buffer.TransactionEntry;
-import org.apache.pulsar.transaction.buffer.exceptions.EndOfTransactionException;
+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.transaction.impl.common.TxnID;
 
 /**
  * A {@link TransactionBufferReader} implementation that reads entries from {@link InMemTransactionBuffer}.
  */
-class InMemTransactionBufferReader implements TransactionBufferReader {
+public class InMemTransactionBufferReader implements TransactionBufferReader {
 
     private final TxnID txnId;
     private final Iterator<Entry<Long, ByteBuf>> entries;
@@ -41,10 +41,8 @@ class InMemTransactionBufferReader implements TransactionBufferReader {
 
     // the iterator should hold the references to the entries
     // so when the reader is closed, all the entries can be released.
-    InMemTransactionBufferReader(TxnID txnId,
-                                 Iterator<Entry<Long, ByteBuf>> entries,
-                                 long committedAtLedgerId,
-                                 long committedAtEntryId) {
+    public InMemTransactionBufferReader(TxnID txnId, Iterator<Entry<Long, ByteBuf>> entries, long committedAtLedgerId,
+                                        long committedAtEntryId) {
         this.txnId = txnId;
         this.entries = entries;
         this.committedAtLedgerId = committedAtLedgerId;

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/impl/PersistentTransactionBuffer.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/PersistentTransactionBuffer.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.impl;
+package org.apache.pulsar.broker.transaction.buffer.impl;
 
 import io.netty.buffer.ByteBuf;
 import java.util.List;
@@ -38,11 +38,11 @@
 import org.apache.pulsar.common.api.proto.PulsarMarkers.MessageIdData;
 import org.apache.pulsar.common.protocol.Markers;
 import org.apache.pulsar.common.util.FutureUtil;
-import org.apache.pulsar.transaction.buffer.TransactionBuffer;
-import org.apache.pulsar.transaction.buffer.TransactionBufferReader;
-import org.apache.pulsar.transaction.buffer.TransactionCursor;
-import org.apache.pulsar.transaction.buffer.TransactionMeta;
-import org.apache.pulsar.transaction.buffer.exceptions.TransactionNotSealedException;
+import org.apache.pulsar.broker.transaction.buffer.TransactionBuffer;
+import org.apache.pulsar.broker.transaction.buffer.TransactionBufferReader;
+import org.apache.pulsar.broker.transaction.buffer.TransactionCursor;
+import org.apache.pulsar.broker.transaction.buffer.TransactionMeta;
+import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotSealedException;
 import org.apache.pulsar.transaction.impl.common.TxnID;
 
 /**

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/impl/PersistentTransactionBufferReader.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/PersistentTransactionBufferReader.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.impl;
+package org.apache.pulsar.broker.transaction.buffer.impl;
 
 import java.util.ArrayList;
 import java.util.Comparator;
@@ -34,10 +34,10 @@
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
 import org.apache.pulsar.common.util.FutureUtil;
-import org.apache.pulsar.transaction.buffer.TransactionBufferReader;
-import org.apache.pulsar.transaction.buffer.TransactionEntry;
-import org.apache.pulsar.transaction.buffer.TransactionMeta;
-import org.apache.pulsar.transaction.buffer.exceptions.TransactionNotSealedException;
+import org.apache.pulsar.broker.transaction.buffer.TransactionBufferReader;
+import org.apache.pulsar.broker.transaction.buffer.TransactionEntry;
+import org.apache.pulsar.broker.transaction.buffer.TransactionMeta;
+import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotSealedException;
 import org.apache.pulsar.transaction.impl.common.TxnStatus;
 
 /**

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/impl/TransactionCursorImpl.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionCursorImpl.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.impl;
+package org.apache.pulsar.broker.transaction.buffer.impl;
 
 import java.util.HashSet;
 import java.util.Map;
@@ -26,10 +26,10 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.pulsar.transaction.buffer.TransactionCursor;
-import org.apache.pulsar.transaction.buffer.TransactionMeta;
-import org.apache.pulsar.transaction.buffer.exceptions.NoTxnsCommittedAtLedgerException;
-import org.apache.pulsar.transaction.buffer.exceptions.TransactionNotFoundException;
+import org.apache.pulsar.broker.transaction.buffer.TransactionCursor;
+import org.apache.pulsar.broker.transaction.buffer.TransactionMeta;
+import org.apache.pulsar.broker.transaction.buffer.exceptions.NoTxnsCommittedAtLedgerException;
+import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotFoundException;
 import org.apache.pulsar.transaction.impl.common.TxnID;
 
 public class TransactionCursorImpl implements TransactionCursor {

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/impl/TransactionEntryImpl.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionEntryImpl.java

@@ -16,24 +16,24 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.impl;
+package org.apache.pulsar.broker.transaction.buffer.impl;
 
 import io.netty.buffer.ByteBuf;
-import org.apache.pulsar.transaction.buffer.TransactionEntry;
+import org.apache.pulsar.broker.transaction.buffer.TransactionEntry;
 import org.apache.pulsar.transaction.impl.common.TxnID;
 
 /**
  * A simple implementation of {@link TransactionEntry}.
  */
-class TransactionEntryImpl implements TransactionEntry {
+public class TransactionEntryImpl implements TransactionEntry {
 
     private final TxnID txnId;
     private final long sequenceId;
     private final long committedAtLedgerId;
     private final long committedAtEntryId;
     private final ByteBuf entryBuf;
 
-    TransactionEntryImpl(TxnID txnId,
+    public TransactionEntryImpl(TxnID txnId,
                          long sequenceId,
                          ByteBuf entryBuf,
                          long committedAtLedgerId,

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/impl/TransactionMetaImpl.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionMetaImpl.java

@@ -16,18 +16,18 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.transaction.buffer.impl;
+package org.apache.pulsar.broker.transaction.buffer.impl;
 
 import com.google.common.annotations.VisibleForTesting;
 import java.util.Map;
 import java.util.SortedMap;
 import java.util.TreeMap;
 import java.util.concurrent.CompletableFuture;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.pulsar.transaction.buffer.TransactionMeta;
-import org.apache.pulsar.transaction.buffer.exceptions.EndOfTransactionException;
-import org.apache.pulsar.transaction.buffer.exceptions.TransactionSealedException;
-import org.apache.pulsar.transaction.buffer.exceptions.UnexpectedTxnStatusException;
+import org.apache.pulsar.broker.transaction.buffer.TransactionMeta;
+import org.apache.pulsar.broker.transaction.buffer.exceptions.EndOfTransactionException;
+import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionSealedException;
+import org.apache.pulsar.broker.transaction.buffer.exceptions.UnexpectedTxnStatusException;
 import org.apache.pulsar.transaction.impl.common.TxnID;
 import org.apache.pulsar.transaction.impl.common.TxnStatus;
 

pulsar-transaction/buffer/src/main/java/org/apache/pulsar/transaction/buffer/impl/package-info.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/package-info.java

@@ -19,4 +19,4 @@
 /**
  * The implementation of a transaction buffer.
  */
-package org.apache.pulsar.transaction.buffer.impl;
+package org.apache.pulsar.broker.transaction.buffer.impl;