Datacouch 1145 transaction support (#1423)

* Porting to SDK-integrated version of transactions

The transactions logic exists in the Java SDK as of 3.3.0,
with a slightly different API.

This is the first effort at the port, which literally just
compiles.  It will not run as crucial code has been
commented and todo-ed.  There is work remaining to figure
out how to complete the port, as some crucial parts (such
as ctx.commit() and ctx.rollback()) have been intentionally
removed.

* Continuing work to get the ExtSDKIntegration port working

Trying to transition to CallbackPreferring manager.

* Added CouchbaseSimpleCallbackTransactionManager, the simplest
possible implementation of CallbackPreferringTransactionManager,
combined with a simpler approach to ThreadLocal storage in
ReactiveInsertByIdSupport.

Test 'commitShouldPersistTxEntriesOfTxAnnotatedMethod' is now
passing.

* Adding WIP get-and-replace @transactional support

(Not yet working as CAS/version field in
Person is not populated correctly.)

Author: programmatix

pom.xml

@@ -21,7 +21,6 @@
         <couchbase>3.2.5</couchbase>
         <couchbase.osgi>3.2.5</couchbase.osgi>
         <springdata.commons>2.7.0-SNAPSHOT</springdata.commons>
-        <couchbase-transactions>1.2.2</couchbase-transactions>
         <java-module-name>spring.data.couchbase</java-module-name>
 <!--
         <spring.boot>2.6.0-RC1</spring.boot>
@@ -56,18 +55,6 @@
 -->
 
 
-    <dependency>
-        <groupId>com.couchbase.client</groupId>
-        <artifactId>couchbase-transactions</artifactId>
-        <version>${couchbase-transactions}</version>
-        <exclusions>
-            <exclusion>
-                <groupId>com.couchbase.client</groupId>
-                <artifactId>java-client</artifactId>
-            </exclusion>
-        </exclusions>
-    </dependency>
-
         <dependency>
             <groupId>org.springframework</groupId>
             <artifactId>spring-context-support</artifactId>
@@ -244,18 +231,6 @@
             <version>4.0.3</version>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>com.couchbase.client</groupId>
-            <artifactId>couchbase-transactions</artifactId>
-            <version>${couchbase-transactions}</version>
-            <scope>compile</scope>
-            <exclusions>
-                <exclusion>
-                    <groupId>com.couchbase.client</groupId>
-                    <artifactId>java-client</artifactId>
-                </exclusion>
-            </exclusions>
-        </dependency>
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>testcontainers</artifactId>
@@ -264,7 +239,7 @@
         <dependency>
             <groupId>com.couchbase.client</groupId>
             <artifactId>java-client</artifactId>
-            <version>3.2.5</version>
+            <version>3.3.0-SNAPSHOT</version>
         </dependency>
 
         <dependency>

src/main/java/com/couchbase/client/java/Cluster.java

@@ -44,6 +44,7 @@
 import com.couchbase.client.java.search.SearchOptions;
 import com.couchbase.client.java.search.SearchQuery;
 import com.couchbase.client.java.search.result.SearchResult;
+import com.couchbase.client.java.transactions.Transactions;
 
 import java.time.Duration;
 import java.util.Map;
@@ -102,6 +103,7 @@
  * The SDK will only work against Couchbase Server 5.0 and later, because RBAC (role-based access control) is a first
  * class concept since 3.0 and therefore required.
  */
+// todo gp is this required?
 public class Cluster implements ClusterInterface {
 
   /**
@@ -574,5 +576,14 @@ public void waitUntilReady(final Duration timeout, final WaitUntilReadyOptions o
     block(asyncCluster.waitUntilReady(timeout, options));
   }
 
+  /**
+   * Allows access to transactions.
+   *
+   * @return the {@link Transactions} interface.
+   */
+  @Stability.Uncommitted
+  public Transactions transactions() {
+    return new Transactions(core(), environment().jsonSerializer());
+  }
 }
 

src/main/java/com/couchbase/transactions/AttemptContextReactiveAccessor.java

@@ -17,74 +17,81 @@
 package com.couchbase.transactions;
 
 import com.couchbase.client.core.annotation.Stability;
-import com.couchbase.transactions.config.MergedTransactionConfig;
-import com.couchbase.transactions.config.PerTransactionConfig;
-import com.couchbase.transactions.config.PerTransactionConfigBuilder;
-import com.couchbase.transactions.config.TransactionConfig;
-import com.couchbase.transactions.forwards.Supported;
-import com.couchbase.transactions.log.TransactionLogger;
+import com.couchbase.client.core.transaction.log.CoreTransactionLogger;
+import com.couchbase.client.java.transactions.ReactiveTransactionAttemptContext;
+import com.couchbase.client.java.transactions.TransactionAttemptContext;
+import org.springframework.transaction.reactive.TransactionContext;
 
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 import java.util.Optional;
 import java.util.UUID;
 
 /**
- * To access the AttemptContextReactive held by AttemptContext
+ * To access the ReactiveTransactionAttemptContext held by TransactionAttemptContext
  *
  * @author Michael Reiche
  */
 public class AttemptContextReactiveAccessor {
 
-  public static AttemptContextReactive getACR(AttemptContext attemptContext) {
-    return attemptContext.ctx();
+  public static ReactiveTransactionAttemptContext getACR(TransactionAttemptContext attemptContext) {
+    // return attemptContext.ctx();
+      // todo gp is this access needed.  Could hold the raw CoreTransactionAttemptContext instead.
+      return null;
   }
 
-  public static AttemptContext from(AttemptContextReactive attemptContextReactive) {
-    return new AttemptContext(attemptContextReactive);
+  public static TransactionAttemptContext from(ReactiveTransactionAttemptContext attemptContextReactive) {
+      // todo gp needed?
+      return null;
+//    return new TransactionAttemptContext(attemptContextReactive);
   }
 
-  public static TransactionLogger getLogger(AttemptContextReactive attemptContextReactive){
-    return attemptContextReactive.LOGGER;
-  }
-  @Stability.Internal
-  public static AttemptContextReactive newAttemptContextReactive(TransactionsReactive transactions){
-    PerTransactionConfig perConfig = PerTransactionConfigBuilder.create().build();
-    MergedTransactionConfig merged = new MergedTransactionConfig(transactions.config(), Optional.of(perConfig));
-
-    TransactionContext overall = new TransactionContext(
-        transactions.cleanup().clusterData().cluster().environment().requestTracer(),
-        transactions.cleanup().clusterData().cluster().environment().eventBus(),
-        UUID.randomUUID().toString(), now(), Duration.ZERO, merged);
-
-    String txnId = UUID.randomUUID().toString();
-    overall.LOGGER.info(configDebug(transactions.config(), perConfig));
-    return transactions.createAttemptContext(overall, merged, txnId);
+  public static CoreTransactionLogger getLogger(ReactiveTransactionAttemptContext attemptContextReactive){
+      // todo gp needed?
+      return null;
+    //return attemptContextReactive;
   }
+    // todo gp needed?
+//  @Stability.Internal
+//  public static ReactiveTransactionAttemptContext newAttemptContextReactive(TransactionsReactive transactions){
+//      return null;
+//    PerTransactionConfig perConfig = PerTransactionConfigBuilder.create().build();
+//    MergedTransactionConfig merged = new MergedTransactionConfig(transactions.config(), Optional.of(perConfig));
+//
+//    TransactionContext overall = new TransactionContext(
+//        transactions.cleanup().clusterData().cluster().environment().requestTracer(),
+//        transactions.cleanup().clusterData().cluster().environment().eventBus(),
+//        UUID.randomUUID().toString(), now(), Duration.ZERO, merged);
+//
+//    String txnId = UUID.randomUUID().toString();
+//    overall.LOGGER.info(configDebug(transactions.config(), perConfig));
+//    return transactions.createAttemptContext(overall, merged, txnId);
+//  }
 
   private static Duration now() {
     return Duration.of(System.nanoTime(), ChronoUnit.NANOS);
   }
 
-  static private String configDebug(TransactionConfig config, PerTransactionConfig perConfig) {
-    StringBuilder sb = new StringBuilder();
-    sb.append("library version: ");
-    sb.append(TransactionsReactive.class.getPackage().getImplementationVersion());
-    sb.append(" config: ");
-    sb.append("atrs=");
-    sb.append(config.numAtrs());
-    sb.append(", metadataCollection=");
-    sb.append(config.metadataCollection());
-    sb.append(", expiry=");
-    sb.append(perConfig.expirationTime().orElse(config.transactionExpirationTime()).toMillis());
-    sb.append("msecs durability=");
-    sb.append(config.durabilityLevel());
-    sb.append(" per-txn config=");
-    sb.append(" durability=");
-    sb.append(perConfig.durabilityLevel());
-    sb.append(", supported=");
-    sb.append(Supported.SUPPORTED);
-    return sb.toString();
-  }
+  // todo gp if needed let's expose in the SDK
+//  static private String configDebug(TransactionConfig config, PerTransactionConfig perConfig) {
+//    StringBuilder sb = new StringBuilder();
+//    sb.append("library version: ");
+//    sb.append(TransactionsReactive.class.getPackage().getImplementationVersion());
+//    sb.append(" config: ");
+//    sb.append("atrs=");
+//    sb.append(config.numAtrs());
+//    sb.append(", metadataCollection=");
+//    sb.append(config.metadataCollection());
+//    sb.append(", expiry=");
+//    sb.append(perConfig.expirationTime().orElse(config.transactionExpirationTime()).toMillis());
+//    sb.append("msecs durability=");
+//    sb.append(config.durabilityLevel());
+//    sb.append(" per-txn config=");
+//    sb.append(" durability=");
+//    sb.append(perConfig.durabilityLevel());
+//    sb.append(", supported=");
+//    sb.append(Supported.SUPPORTED);
+//    return sb.toString();
+//  }
 
 }