Programmatix datacouch 1145 transaction support (#1446)

* 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.)

* Transitioning to use CoreTransactionAttemptContext.
Tests may fail.

* Removing AttemptContextReactiveAccessor

Don't think we need this, as we can pass around
CoreTransactionAttemptContext instead, which gives
access to a lot of internals.

* Removing TransactionsReactive

Would prefer not to C&P a huge class out of the transaction
internals, and don't think we need it.

* Removing some files not currently used

To reduce & simplify the amount of code to look at.

Some don't seem to be used in any branch, some just
aren't used in this branch.

* Removing CouchbaseTransactionInterceptor

As per offline discussion, CallbackPreferringPlatformTransactionManager
is perhaps the optimal solution.

* Copying @transactional tests out into separate class

* Tidyup

* Tidyup test names

* Verify GenericSupport is on same thread before and after transactional operation

* Refactoring CouchbaseSimpleCallbackTransactionManager ThreadLocalStorage management

* Using latest java-client

* ReactiveReplaceByIdSupport - Fixing use of CAS now have CoreTransactionAttemptContext.

Removing unused code.

* ReactiveInsertByIdSupport - fixing use of reactive vs non-reactive, and CAS

* Merging upstream

* Remove incorrect thread check (.doOnNext could execute on a different thread)

* Completing merge from upstream

* Removing unused classes

* Give GenericSupport a better name

* Reject at runtime options that aren't supported in a transaction

* Fixing some small todos, partly by removing unused coe

* Fix runtime option checks

* Simplifying CouchbaseSimpleCallbackTransactionManager ThreadLocalStorage

Standardising on ReactiveCouchbaseResourceHolder rather than
holding CoreTransactionAttemptContext too

* Removing version from CouchbaseDocument

Can't recall why I added this, and tests pass without it

* Improving CouchbaseTransactionalIntegrationTests and adding more tests

* Reject operations that aren't allowed in a transaction (upsertById etc.)

* Improve handling of CAS mismatch

By calling CoreTransactionAttemptContext.operationFailed,
it ensures that internal state is set.  So even if
the user catches the exception, the transaction still behaves
as it should.

* Removing a now-redundant non-transactional check on upsertById

I missed this when adding TransactionalSupport.verifyNotInTransaction
here.

* Support @transactional options timeout and isolation level

Co-authored-by: Graham Pople <[email protected]>

Author: mikereiche

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

@@ -103,7 +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?
+// todo gpx as per discussion with miker - if required, ClusterInterface will be added to the SDK instead
 public class Cluster implements ClusterInterface {
 
   /**
@@ -256,9 +256,9 @@ public static Cluster connect(final String connectionString, final ClusterOption
     final ClusterOptions.Built opts = options.build();
     final Supplier<ClusterEnvironment> environmentSupplier = extractClusterEnvironment(connectionString, opts);
     return new Cluster(
-      environmentSupplier,
-      opts.authenticator(),
-      seedNodesFromConnectionString(connectionString, environmentSupplier.get())
+            environmentSupplier,
+            opts.authenticator(),
+            seedNodesFromConnectionString(connectionString, environmentSupplier.get())
     );
   }
 

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

@@ -83,7 +83,7 @@ public interface ClusterInterface {
 
   //AnalyticsResult analyticsQuery(String statement);
 
- // AnalyticsResult analyticsQuery(String statement, AnalyticsOptions options);
+  // AnalyticsResult analyticsQuery(String statement, AnalyticsOptions options);
 
   SearchResult searchQuery(String indexName, SearchQuery query);
 

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

@@ -44,12 +44,6 @@
  */
 public class AttemptContextReactiveAccessor {
 
-	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 ReactiveTransactions reactive(Transactions transactions) {
 		try {
 			Field field = Transactions.class.getDeclaredField("reactive");
@@ -212,26 +206,18 @@ public static TransactionResult run(Transactions transactions, Consumer<Transact
 		return reactive(transactions).runBlocking(transactionLogic, coreTransactionOptions);
 	}
 
-	// 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();
-	// }
+		CoreTransactionAttemptContext coreTransactionsReactive;
+		try {
+			Field field = TransactionAttemptContext.class.getDeclaredField("internal");
+			field.setAccessible(true);
+			coreTransactionsReactive = (CoreTransactionAttemptContext) field.get(atr);
+		} catch (Throwable err) {
+			throw new RuntimeException(err);
+		}
+		return coreTransactionsReactive;
+	}
 
+	public static ReactiveTransactionAttemptContext createReactiveTransactionAttemptContext(CoreTransactionAttemptContext core, JsonSerializer jsonSerializer) {
+		return new ReactiveTransactionAttemptContext(core, jsonSerializer);
+	}
 }

src/main/java/com/example/demo/CouchbaseTransactionalTemplate.java

@@ -26,8 +26,7 @@ public <T> SpringTransactionGetResult<T> findById(String id, Class<T> domainType
 			CoreTransactionAttemptContext ctx = getContext();
 			CoreTransactionGetResult getResult = ctx.get(  makeCollectionIdentifier(template.getCouchbaseClientFactory().getDefaultCollection().async()) , id).block();
 
-			// todo gp getResult.cas() is no longer exposed - required?
-			T t = template.support().decodeEntity(id, new String(getResult.contentAsBytes()), 0, domainType,
+			T t = template.support().decodeEntity(id, new String(getResult.contentAsBytes()), getResult.cas(), domainType,
 					null, null, null);
 			return new SpringTransactionGetResult<>(t, getResult);
 		} catch (Exception e) {

src/main/java/com/example/demo/SpringTransactionGetResult.java

@@ -24,8 +24,8 @@ public CoreTransactionGetResult getInner() {
   @Override
   public String toString() {
     return "SpringTransactionGetResult{" +
-      "value=" + value +
-      ", inner=" + inner +
-      '}';
+            "value=" + value +
+            ", inner=" + inner +
+            '}';
   }
 }

src/main/java/org/springframework/data/couchbase/CouchbaseClientFactory.java

@@ -77,7 +77,7 @@ public interface CouchbaseClientFactory extends Closeable {
 	PersistenceExceptionTranslator getExceptionTranslator();
 
 	CoreTransactionAttemptContext getCore(TransactionOptions options,
-																				CoreTransactionAttemptContext atr);
+										  CoreTransactionAttemptContext atr);
 
 	//CouchbaseClientFactory with(CouchbaseTransactionalOperator txOp);
 

src/main/java/org/springframework/data/couchbase/ReactiveCouchbaseClientFactory.java

@@ -23,7 +23,6 @@
 import com.couchbase.client.java.Scope;
 import com.couchbase.client.java.transactions.ReactiveTransactionAttemptContext;
 import com.couchbase.client.java.transactions.config.TransactionOptions;
-import org.springframework.data.couchbase.transaction.ClientSessionOptions;
 import org.springframework.data.couchbase.transaction.CouchbaseTransactionalOperator;
 import org.springframework.data.couchbase.transaction.ReactiveCouchbaseResourceHolder;
 import reactor.core.publisher.Mono;
@@ -107,8 +106,7 @@ public interface ReactiveCouchbaseClientFactory /*extends CodecRegistryProvider*
 
 	void close() throws IOException;
 
-	ReactiveCouchbaseResourceHolder getTransactionResources(TransactionOptions options,
-			CoreTransactionAttemptContext ctx);
+	ReactiveCouchbaseResourceHolder getTransactionResources(TransactionOptions options, CoreTransactionAttemptContext ctx);
 
 	/*
 	 * (non-Javadoc)

src/main/java/org/springframework/data/couchbase/SimpleCouchbaseClientFactory.java

@@ -53,21 +53,21 @@ public class SimpleCouchbaseClientFactory implements CouchbaseClientFactory {
 	//private JsonSerializer serializer = null;
 
 	public SimpleCouchbaseClientFactory(final String connectionString, final Authenticator authenticator,
-			final String bucketName) {
+										final String bucketName) {
 		this(connectionString, authenticator, bucketName, null);
 	}
 
 	public SimpleCouchbaseClientFactory(final String connectionString, final Authenticator authenticator,
-			final String bucketName, final String scopeName) {
+										final String bucketName, final String scopeName) {
 		this(new OwnedSupplier<>(Cluster.connect(connectionString, ClusterOptions.clusterOptions(authenticator)
-				// todo gp disabling cleanupLostAttempts to simplify output during development
-				.environment(env -> env.transactionsConfig(
-						TransactionsConfig.cleanupConfig(TransactionsCleanupConfig.cleanupLostAttempts(false)))))),
+						// todo gp disabling cleanupLostAttempts to simplify output during development
+						.environment(env -> env.transactionsConfig(
+								TransactionsConfig.cleanupConfig(TransactionsCleanupConfig.cleanupLostAttempts(false)))))),
 				bucketName, scopeName);
 	}
 
 	public SimpleCouchbaseClientFactory(final String connectionString, final Authenticator authenticator,
-			final String bucketName, final String scopeName, final ClusterEnvironment environment) {
+										final String bucketName, final String scopeName, final ClusterEnvironment environment) {
 		this(
 				new OwnedSupplier<>(
 						Cluster.connect(connectionString, ClusterOptions.clusterOptions(authenticator).environment(environment))),
@@ -81,7 +81,7 @@ public SimpleCouchbaseClientFactory(final Cluster cluster, final String bucketNa
 	}
 
 	private SimpleCouchbaseClientFactory(final Supplier<Cluster> cluster, final String bucketName,
-			final String scopeName) {
+										 final String scopeName) {
 		this.cluster = cluster;
 		this.bucket = cluster.get().bucket(bucketName);
 		this.scope = scopeName == null ? bucket.defaultScope() : bucket.scope(scopeName);

src/main/java/org/springframework/data/couchbase/SimpleReactiveCouchbaseClientFactory.java

@@ -38,7 +38,7 @@ public class SimpleReactiveCouchbaseClientFactory implements ReactiveCouchbaseCl
 	CouchbaseTransactionalOperator transactionalOperator;
 
 	public SimpleReactiveCouchbaseClientFactory(Cluster cluster, String bucketName, String scopeName,
-			CouchbaseTransactionalOperator transactionalOperator) {
+												CouchbaseTransactionalOperator transactionalOperator) {
 		this.cluster = Mono.just(cluster);
 		this.theCluster = cluster;
 		this.bucketName = bucketName;
@@ -146,15 +146,13 @@ public void close() {
 	}
 
 	@Override
-	public Mono<ReactiveCouchbaseResourceHolder> getTransactionResources(TransactionOptions options) { // hopefully this
-																																																			// gets filled in
-																																																			// later
-		return Mono.just(new ReactiveCouchbaseResourceHolder(null));
+	public Mono<ReactiveCouchbaseResourceHolder> getTransactionResources(TransactionOptions options) { 
+	return Mono.just(new ReactiveCouchbaseResourceHolder(null));
 	}
 
 	@Override
 	public ReactiveCouchbaseResourceHolder getTransactionResources(TransactionOptions options,
-			CoreTransactionAttemptContext atr) {
+																   CoreTransactionAttemptContext atr) {
 		if (atr == null) {
 			atr = AttemptContextReactiveAccessor
 					.newCoreTranactionAttemptContext(AttemptContextReactiveAccessor.reactive(transactions));
@@ -218,7 +216,7 @@ static final class CoreTransactionAttemptContextBoundCouchbaseClientFactory
 		// private final Transactions transactions;
 
 		CoreTransactionAttemptContextBoundCouchbaseClientFactory(ReactiveCouchbaseResourceHolder transactionResources,
-				ReactiveCouchbaseClientFactory delegate, Transactions transactions) {
+																 ReactiveCouchbaseClientFactory delegate, Transactions transactions) {
 			this.transactionResources = transactionResources;
 			this.delegate = delegate;
 			// this.transactions = transactions;
@@ -308,7 +306,7 @@ public Mono<ReactiveCouchbaseResourceHolder> getTransactionResources(Transaction
 
 		@Override
 		public ReactiveCouchbaseResourceHolder getTransactionResources(TransactionOptions options,
-				CoreTransactionAttemptContext atr) {
+																	   CoreTransactionAttemptContext atr) {
 			ReactiveCouchbaseResourceHolder holder = delegate.getTransactionResources(options, atr);
 			return holder;
 		}

src/main/java/org/springframework/data/couchbase/config/AbstractCouchbaseConfiguration.java

@@ -168,6 +168,7 @@ public ClusterEnvironment couchbaseClusterEnvironment() {
 			throw new CouchbaseException("non-shadowed Jackson not present");
 		}
 		builder.jsonSerializer(JacksonJsonSerializer.create(couchbaseObjectMapper()));
+		// todo gp only suitable for tests
 		TransactionsConfig.cleanupConfig(TransactionsCleanupConfig.cleanupLostAttempts(false));
 		builder.transactionsConfig(transactionsConfig());
 		configureEnvironment(builder);
@@ -185,15 +186,15 @@ protected void configureEnvironment(final ClusterEnvironment.Builder builder) {
 
 	@Bean(name = BeanNames.COUCHBASE_TEMPLATE)
 	public CouchbaseTemplate couchbaseTemplate(CouchbaseClientFactory couchbaseClientFactory,
-			ReactiveCouchbaseClientFactory reactiveCouchbaseClientFactory,
-			MappingCouchbaseConverter mappingCouchbaseConverter, TranslationService couchbaseTranslationService) {
+											   ReactiveCouchbaseClientFactory reactiveCouchbaseClientFactory,
+											   MappingCouchbaseConverter mappingCouchbaseConverter, TranslationService couchbaseTranslationService) {
 		return new CouchbaseTemplate(couchbaseClientFactory, reactiveCouchbaseClientFactory, mappingCouchbaseConverter,
 				couchbaseTranslationService, getDefaultConsistency());
 	}
 
 	public CouchbaseTemplate couchbaseTemplate(CouchbaseClientFactory couchbaseClientFactory,
-			ReactiveCouchbaseClientFactory reactiveCouchbaseClientFactory,
-			MappingCouchbaseConverter mappingCouchbaseConverter) {
+											   ReactiveCouchbaseClientFactory reactiveCouchbaseClientFactory,
+											   MappingCouchbaseConverter mappingCouchbaseConverter) {
 		return couchbaseTemplate(couchbaseClientFactory, reactiveCouchbaseClientFactory, mappingCouchbaseConverter,
 				new JacksonTranslationService());
 	}
@@ -291,7 +292,7 @@ public String typeKey() {
 	 */
 	@Bean
 	public MappingCouchbaseConverter mappingCouchbaseConverter(CouchbaseMappingContext couchbaseMappingContext,
-			CouchbaseCustomConversions couchbaseCustomConversions) {
+															   CouchbaseCustomConversions couchbaseCustomConversions) {
 		MappingCouchbaseConverter converter = new MappingCouchbaseConverter(couchbaseMappingContext, typeKey());
 		converter.setCustomConversions(couchbaseCustomConversions);
 		return converter;
@@ -346,12 +347,6 @@ public ObjectMapper couchbaseObjectMapper() {
 
 	/*****  ALL THIS TX SHOULD BE MOVED OUT INTO THE IMPL OF AbstractCouchbaseConfiguration *****/
 
-	// todo gp how to DI this into the Cluster creation esp. as it creates a CoreTransactionConfig
-//	@Bean
-//	public TransactionsConfig transactionConfig() {
-//		return TransactionsConfig.builder().build();
-//	}
-
 	@Bean(BeanNames.REACTIVE_COUCHBASE_TRANSACTION_MANAGER)
 	ReactiveCouchbaseTransactionManager reactiveTransactionManager(
 			ReactiveCouchbaseClientFactory reactiveCouchbaseClientFactory) {
@@ -377,11 +372,13 @@ CouchbaseTransactionManager transactionManager(CouchbaseClientFactory clientFact
 		return new CouchbaseTransactionManager(clientFactory, options);
 	}
 
+	// todo gpx these would be per-transactions options so it seems odd to have a global bean?  Surely would want to configure everything at global level instead?
 	@Bean
 	public TransactionOptions transactionsOptions(){
 		return TransactionOptions.transactionOptions();
 	}
 
+	// todo gpx transactions config is now done in standard ClusterConfig - so I think we don't want a separate bean?
 	public TransactionsConfig.Builder transactionsConfig(){
 		return TransactionsConfig.builder().durabilityLevel(DurabilityLevel.NONE).timeout(Duration.ofMinutes(20));// for testing
 	}