GEODE-6641: Code impacts for support for two phase commit transactions

clicache/src/Cache.cpp

@@ -99,8 +99,7 @@ namespace Apache
         // TODO shared_ptr this should be checking the return type for which tx mgr
         try
         {
-          auto nativeptr = std::dynamic_pointer_cast<InternalCacheTransactionManager2PC>(
-            m_nativeptr->get()->getCacheTransactionManager());
+          auto nativeptr = m_nativeptr->get()->getCacheTransactionManager();
           return Apache::Geode::Client::CacheTransactionManager::Create(nativeptr.get());
         }
         finally

clicache/src/CacheTransactionManager.hpp

@@ -21,7 +21,6 @@
 #include "geode_defs.hpp"
 #include "begin_native.hpp"
 #include <geode/CacheTransactionManager.hpp>
-#include "InternalCacheTransactionManager2PC.hpp"
 #include "end_native.hpp"
 #include "native_shared_ptr.hpp"
 #include "TransactionId.hpp"
@@ -204,7 +203,7 @@ namespace Apache
 
       internal:
 
-        inline static CacheTransactionManager^ Create(native::InternalCacheTransactionManager2PC* nativeptr )
+        inline static CacheTransactionManager^ Create(native::CacheTransactionManager* nativeptr )
         {
           return ( nativeptr != nullptr ?
             gcnew CacheTransactionManager( nativeptr ) : nullptr );
@@ -217,12 +216,12 @@ namespace Apache
         /// Private constructor to wrap a native object pointer
         /// </summary>
         /// <param name="nativeptr">The native object pointer</param>
-        inline CacheTransactionManager(native::InternalCacheTransactionManager2PC* nativeptr )
+        inline CacheTransactionManager(native::CacheTransactionManager* nativeptr )
           : m_nativeptr(nativeptr)
         {
         }
 
-        native::InternalCacheTransactionManager2PC* m_nativeptr;
+        native::CacheTransactionManager* m_nativeptr;
       };
     }  // namespace Client
   }  // namespace Geode

cppcache/include/geode/CacheTransactionManager.hpp

@@ -47,6 +47,38 @@ class APACHE_GEODE_EXPORT CacheTransactionManager {
    */
   virtual void begin() = 0;
 
+  /**
+   * Performs prepare during 2 phase commit completion, for the transaction
+   * associated with the current thread.
+   * Locks of the entries modified in the current transaction on the server
+   * side. If the prepare operation fails due to a conflict it will destroy
+   * the transaction state and throw a {@link CommitConflictException}.
+   * If the prepare operation succeeds, transaction state is set to
+   * prepared state.  When this method completes, the thread is still
+   * associated with a transaction, and is waiting on commit or rollback
+   * operation.
+   *
+   * @throws IllegalStateException if the thread is not associated with a
+   * transaction
+   *
+   * @throws CommitConflictException if the commit operation fails due to
+   * a write conflict.
+   *
+   * @throws TransactionDataNodeHasDepartedException if the node hosting the
+   * transaction data has departed. This is only relevant for transaction that
+   * involve PartitionedRegions.
+   *
+   * @throws TransactionDataNotColocatedException if at commit time, the data
+   * involved in the transaction has moved away from the transaction hosting
+   * node. This can only happen if rebalancing/recovery happens during a
+   * transaction that involves a PartitionedRegion.
+   *
+   * @throws TransactionInDoubtException when Geode cannot tell which nodes
+   * have applied the transaction and which have not. This only occurs if nodes
+   * fail mid-commit, and only then in very rare circumstances.
+   */
+  virtual void prepare() = 0;
+
   /** Commit the transaction associated with the current thread. If
    *  the commit operation fails due to a conflict it will destroy
    *  the transaction state and throw a {@link

cppcache/integration-test/ThinClientTransactions.hpp

@@ -1002,6 +1002,7 @@ DUNIT_TASK_DEFINITION(CLIENT2, UpdateClient2Entries)
     LOG("StepSix complete.");
   }
 END_TASK_DEFINITION
+
 DUNIT_TASK_DEFINITION(CLIENT1, CreateClient1EntryTwice)
   { createEntryTwice(regionNames[0], CREATE_TWICE_KEY, CREATE_TWICE_VALUE); }
 END_TASK_DEFINITION

cppcache/integration-test/ThinClientTransactionsXA.hpp

@@ -27,7 +27,7 @@
 
 #include <string>
 #include <geode/TransactionId.hpp>
-#include <InternalCacheTransactionManager2PC.hpp>
+#include <geode/CacheTransactionManager.hpp>
 
 #define ROOT_NAME "ThinClientTransactionsXA"
 #define ROOT_SCOPE DISTRIBUTED_ACK
@@ -43,7 +43,7 @@ using apache::geode::client::CacheServerException;
 using apache::geode::client::EntryExistsException;
 using apache::geode::client::EntryNotFoundException;
 using apache::geode::client::IllegalStateException;
-using apache::geode::client::InternalCacheTransactionManager2PC;
+using apache::geode::client::CacheTransactionManager;
 using apache::geode::client::Properties;
 using apache::geode::client::TransactionException;
 using apache::geode::client::TransactionId;
@@ -384,9 +384,7 @@ class SuspendTransactionThread : public ACE_Task_Base {
     sprintf(buf, " In SuspendTransactionThread");
     LOG(buf);
 
-    auto txManager =
-        std::dynamic_pointer_cast<InternalCacheTransactionManager2PC>(
-            getHelper()->getCache()->getCacheTransactionManager());
+    auto txManager = getHelper()->getCache()->getCacheTransactionManager();
 
     txManager->begin();
 
@@ -462,9 +460,7 @@ class ResumeTransactionThread : public ACE_Task_Base {
                      "In ResumeTransactionThread - Key should not have been "
                      "found in region.");
 
-    auto txManager =
-        std::dynamic_pointer_cast<InternalCacheTransactionManager2PC>(
-            getHelper()->getCache()->getCacheTransactionManager());
+    auto txManager = getHelper()->getCache()->getCacheTransactionManager();
     if (m_tryResumeWithSleep) {
       THREADERRORCHECK(!txManager->isSuspended(m_suspendedTransaction),
                        "In ResumeTransactionThread - the transaction should "
@@ -580,9 +576,7 @@ END_TASK_DEFINITION
 
 DUNIT_TASK_DEFINITION(CLIENT1, SuspendResumeCommit)
   {
-    auto txManager =
-        std::dynamic_pointer_cast<InternalCacheTransactionManager2PC>(
-            getHelper()->getCache()->getCacheTransactionManager());
+    auto txManager = getHelper()->getCache()->getCacheTransactionManager();
     auto regPtr0 = getHelper()->getRegion(regionNames[0]);
     ASSERT(regPtr0 != nullptr, "In SuspendResumeCommit - Region not found.");
     auto regPtr1 = getHelper()->getRegion(regionNames[1]);
@@ -665,9 +659,7 @@ END_TASK_DEFINITION
 
 DUNIT_TASK_DEFINITION(CLIENT1, SuspendTimeOut)
   {
-    auto txManager =
-        std::dynamic_pointer_cast<InternalCacheTransactionManager2PC>(
-            getHelper()->getCache()->getCacheTransactionManager());
+    auto txManager = getHelper()->getCache()->getCacheTransactionManager();
     auto keyPtr4 = CacheableKey::create(keys[4]);
     auto keyPtr5 = CacheableKey::create(keys[5]);
 
@@ -709,9 +701,7 @@ END_TASK_DEFINITION
 
 DUNIT_TASK_DEFINITION(CLIENT1, SuspendResumeRollback)
   {
-    auto txManager =
-        std::dynamic_pointer_cast<InternalCacheTransactionManager2PC>(
-            getHelper()->getCache()->getCacheTransactionManager());
+    auto txManager = getHelper()->getCache()->getCacheTransactionManager();
     auto keyPtr4 = CacheableKey::create(keys[4]);
     auto keyPtr5 = CacheableKey::create(keys[5]);
     auto keyPtr6 = CacheableKey::create(keys[6]);
@@ -951,9 +941,7 @@ END_TASK_DEFINITION
 
 DUNIT_TASK_DEFINITION(CLIENT1, StepThree)
   {
-    auto txManager =
-        std::dynamic_pointer_cast<InternalCacheTransactionManager2PC>(
-            getHelper()->getCache()->getCacheTransactionManager());
+    auto txManager = getHelper()->getCache()->getCacheTransactionManager();
     txManager->begin();
     createEntry(regionNames[0], keys[0], vals[0]);
     createEntry(regionNames[1], keys[2], vals[2]);
@@ -967,9 +955,7 @@ DUNIT_TASK_DEFINITION(CLIENT2, StepFour)
   {
     doNetsearch(regionNames[0], keys[0], vals[0]);
     doNetsearch(regionNames[1], keys[2], vals[2]);
-    auto txManager =
-        std::dynamic_pointer_cast<InternalCacheTransactionManager2PC>(
-            getHelper()->getCache()->getCacheTransactionManager());
+    auto txManager = getHelper()->getCache()->getCacheTransactionManager();
     txManager->begin();
     createEntry(regionNames[0], keys[1], vals[1]);
     createEntry(regionNames[1], keys[3], vals[3]);
@@ -1016,9 +1002,7 @@ DUNIT_TASK_DEFINITION(CLIENT2, StepSix)
   {
     doNetsearch(regionNames[0], keys[0], vals[0]);
     doNetsearch(regionNames[1], keys[2], vals[2]);
-    auto txManager =
-        std::dynamic_pointer_cast<InternalCacheTransactionManager2PC>(
-            getHelper()->getCache()->getCacheTransactionManager());
+    auto txManager = getHelper()->getCache()->getCacheTransactionManager();
     txManager->begin();
     updateEntry(regionNames[0], keys[1], nvals[1]);
     updateEntry(regionNames[1], keys[3], nvals[3]);

cppcache/integration/test/ClientTransactionXATest.cpp

@@ -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.
+ */
+
+#include <gmock/gmock.h>
+
+#include <future>
+#include <thread>
+
+#include <gtest/gtest.h>
+
+#include <geode/Cache.hpp>
+#include <geode/CacheTransactionManager.hpp>
+#include <geode/RegionFactory.hpp>
+#include <geode/RegionShortcut.hpp>
+
+#include "framework/Cluster.h"
+
+namespace {
+
+using apache::geode::client::CacheableString;
+
+std::shared_ptr<apache::geode::client::Region> setupRegion(
+    apache::geode::client::Cache &cache) {
+  auto region =
+      cache.createRegionFactory(apache::geode::client::RegionShortcut::PROXY)
+          .setPoolName("default")
+          .create("region");
+  return region;
+}
+
+TEST(ClientTransactionXATest, interTxand2PCTx) {
+  Cluster cluster{LocatorCount{1}, ServerCount{1}};
+  cluster.getGfsh()
+      .create()
+      .region()
+      .withName("region")
+      .withType("PARTITION")
+      .execute();
+
+  auto cache = cluster.createCache();
+  auto region = setupRegion(cache);
+
+  cache.getCacheTransactionManager()->begin();
+  region->put("one", "one");
+  cache.getCacheTransactionManager()->commit();
+  auto v1 = std::dynamic_pointer_cast<CacheableString>(region->get("one"));
+  EXPECT_EQ("one", v1->value());
+
+  cache.getCacheTransactionManager()->begin();
+  region->put("two", "two");
+  cache.getCacheTransactionManager()->prepare();
+  cache.getCacheTransactionManager()->commit();
+  auto v2 = std::dynamic_pointer_cast<CacheableString>(region->get("two"));
+  EXPECT_EQ("two", v2->value());
+
+  cache.getCacheTransactionManager()->begin();
+  region->put("two", "three");
+  cache.getCacheTransactionManager()->prepare();
+  cache.getCacheTransactionManager()->rollback();
+  auto v3 = std::dynamic_pointer_cast<CacheableString>(region->get("two"));
+  EXPECT_EQ("two", v3->value());
+
+  cache.getCacheTransactionManager()->begin();
+  region->put("one", "three");
+  cache.getCacheTransactionManager()->rollback();
+  auto v4 = std::dynamic_pointer_cast<CacheableString>(region->get("one"));
+  EXPECT_EQ("one", v4->value());
+
+  cache.getCacheTransactionManager()->begin();
+  region->put("one", "two");
+  cache.getCacheTransactionManager()->prepare();
+  cache.getCacheTransactionManager()->commit();
+  auto v5 = std::dynamic_pointer_cast<CacheableString>(region->get("one"));
+  EXPECT_EQ("two", v5->value());
+}
+
+}  // namespace

cppcache/src/CacheImpl.cpp

@@ -79,7 +79,7 @@ CacheImpl::CacheImpl(Cache* c, const std::shared_ptr<Properties>& dsProps,
       m_authInitialize(authInitialize) {
   using apache::geode::statistics::StatisticsManager;
 
-  m_cacheTXManager = std::shared_ptr<InternalCacheTransactionManager2PC>(
+  m_cacheTXManager = std::shared_ptr<CacheTransactionManager>(
       new InternalCacheTransactionManager2PCImpl(this));
 
   auto& prop = m_distributedSystem.getSystemProperties();