TEPHRA-179 Transaction service high availability changes

tephra-core/src/main/java/org/apache/tephra/TransactionManager.java

@@ -18,6 +18,7 @@
 
 package org.apache.tephra;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Objects;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Stopwatch;
@@ -1265,6 +1266,11 @@ public void logStatistics() {
                ", committed = " + committedChangeSets.size());
   }
 
+  @VisibleForTesting
+  public TransactionStateStorage getTransactionStateStorage() {
+    return persistor;
+  }
+
   private abstract static class DaemonThreadExecutor extends Thread {
     private AtomicBoolean stopped = new AtomicBoolean(false);
 

tephra-core/src/main/java/org/apache/tephra/TransactionServiceMain.java

@@ -24,8 +24,7 @@
 import org.apache.tephra.distributed.TransactionService;
 import org.apache.tephra.runtime.ConfigModule;
 import org.apache.tephra.runtime.DiscoveryModules;
-import org.apache.tephra.runtime.TransactionClientModule;
-import org.apache.tephra.runtime.TransactionModules;
+import org.apache.tephra.runtime.TransactionServiceModule;
 import org.apache.tephra.runtime.ZKModule;
 import org.apache.tephra.util.ConfigurationFactory;
 import org.apache.twill.zookeeper.ZKClientService;
@@ -104,8 +103,7 @@ public void start() throws Exception {
       new ConfigModule(conf),
       new ZKModule(),
       new DiscoveryModules().getDistributedModules(),
-      new TransactionModules().getDistributedModules(),
-      new TransactionClientModule()
+      new TransactionServiceModule()
     );
 
     ZKClientService zkClientService = injector.getInstance(ZKClientService.class);

tephra-core/src/main/java/org/apache/tephra/distributed/TransactionService.java

@@ -19,16 +19,19 @@
 package org.apache.tephra.distributed;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.AbstractService;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.Uninterruptibles;
 import com.google.inject.Inject;
 import com.google.inject.Provider;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.tephra.TransactionManager;
+import org.apache.tephra.TxConstants;
 import org.apache.tephra.distributed.thrift.TTransactionServer;
-import org.apache.tephra.inmemory.InMemoryTransactionService;
 import org.apache.tephra.rpc.ThriftRPCServer;
 import org.apache.twill.api.ElectionHandler;
+import org.apache.twill.common.Cancellable;
+import org.apache.twill.discovery.Discoverable;
 import org.apache.twill.discovery.DiscoveryService;
 import org.apache.twill.internal.ServiceListenerAdapter;
 import org.apache.twill.internal.zookeeper.LeaderElection;
@@ -42,28 +45,64 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import javax.annotation.Nullable;
 
 /**
  *
  */
-public final class TransactionService extends InMemoryTransactionService {
+public class TransactionService extends AbstractService {
   private static final Logger LOG = LoggerFactory.getLogger(TransactionService.class);
   private LeaderElection leaderElection;
   private final ZKClient zkClient;
+  private final DiscoveryService discoveryService;
+  private final Provider<TransactionManager> txManagerProvider;
 
+  private final String serviceName;
+  private Cancellable cancelDiscovery;
+
+  // thrift server config
+  private final String address;
+  private final int port;
+  private final int threads;
+  private final int ioThreads;
+  private final int maxReadBufferBytes;
+
+  private TransactionManager txManager;
   private ThriftRPCServer<TransactionServiceThriftHandler, TTransactionServer> server;
 
   @Inject
   public TransactionService(Configuration conf,
                             ZKClient zkClient,
                             DiscoveryService discoveryService,
                             Provider<TransactionManager> txManagerProvider) {
-    super(conf, discoveryService, txManagerProvider);
     this.zkClient = zkClient;
+    this.discoveryService = discoveryService;
+    this.txManagerProvider = txManagerProvider;
+
+    serviceName = conf.get(TxConstants.Service.CFG_DATA_TX_DISCOVERY_SERVICE_NAME,
+                           TxConstants.Service.DEFAULT_DATA_TX_DISCOVERY_SERVICE_NAME);
+
+    address = conf.get(TxConstants.Service.CFG_DATA_TX_BIND_ADDRESS, TxConstants.Service.DEFAULT_DATA_TX_BIND_ADDRESS);
+    port = conf.getInt(TxConstants.Service.CFG_DATA_TX_BIND_PORT, TxConstants.Service.DEFAULT_DATA_TX_BIND_PORT);
+
+    // Retrieve the number of threads for the service
+    threads = conf.getInt(TxConstants.Service.CFG_DATA_TX_SERVER_THREADS,
+                          TxConstants.Service.DEFAULT_DATA_TX_SERVER_THREADS);
+    ioThreads = conf.getInt(TxConstants.Service.CFG_DATA_TX_SERVER_IO_THREADS,
+                            TxConstants.Service.DEFAULT_DATA_TX_SERVER_IO_THREADS);
+
+    maxReadBufferBytes = conf.getInt(TxConstants.Service.CFG_DATA_TX_THRIFT_MAX_READ_BUFFER,
+                                     TxConstants.Service.DEFAULT_DATA_TX_THRIFT_MAX_READ_BUFFER);
+
+    LOG.info("Configuring TransactionService" +
+               ", address: " + address +
+               ", port: " + port +
+               ", threads: " + threads +
+               ", io threads: " + ioThreads +
+               ", max read buffer (bytes): " + maxReadBufferBytes);
   }
 
-  @Override
-  protected InetSocketAddress getAddress() {
+  private InetSocketAddress getAddress() {
     if (address.equals("0.0.0.0")) {
       // resolve hostname
       try {
@@ -112,6 +151,7 @@ public void failed(State from, Throwable failure) {
       public void follower() {
         // First stop the transaction server as un-registering from discovery can block sometimes.
         // That can lead to multiple transaction servers being active at the same time.
+        txManager = null;
         if (server != null && server.isRunning()) {
           server.stopAndWait();
         }
@@ -140,7 +180,7 @@ protected void abort(Throwable cause) {
     notifyFailed(cause);
   }
 
-  protected void internalStop() {
+  private void internalStop() {
     if (leaderElection != null) {
       // NOTE: if was a leader this will cause loosing of leadership which in callback above will
       //       de-register service in discovery service and stop the service if needed
@@ -153,4 +193,31 @@ protected void internalStop() {
       }
     }
   }
+
+  private void undoRegister() {
+    if (cancelDiscovery != null) {
+      cancelDiscovery.cancel();
+    }
+  }
+
+  private void doRegister() {
+    cancelDiscovery = discoveryService.register(new Discoverable() {
+      @Override
+      public String getName() {
+        return serviceName;
+      }
+
+      @Override
+      public InetSocketAddress getSocketAddress() {
+        return getAddress();
+      }
+    });
+  }
+
+  @SuppressWarnings("WeakerAccess")
+  @VisibleForTesting
+  @Nullable
+  public TransactionManager getTransactionManager() {
+    return txManager;
+  }
 }

tephra-core/src/main/java/org/apache/tephra/runtime/DefaultTransactionManagerProvider.java

@@ -0,0 +1,71 @@
+/*
+ * 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.tephra.runtime;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Inject;
+import com.google.inject.Injector;
+import com.google.inject.Module;
+import com.google.inject.Provider;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.tephra.TransactionManager;
+import org.apache.twill.zookeeper.ZKClient;
+import org.apache.twill.zookeeper.ZKClientService;
+
+/**
+ * A provider for {@link TransactionManager} that provides a new instance every time.
+ */
+public class DefaultTransactionManagerProvider implements Provider<TransactionManager> {
+  private final Configuration conf;
+  private final ZKClientService zkClientService;
+
+  @Inject
+  public DefaultTransactionManagerProvider(Configuration conf, ZKClientService zkClientService) {
+    this.conf = conf;
+    this.zkClientService = zkClientService;
+  }
+
+  @Override
+  public TransactionManager get() {
+    // Create a new injector every time since Guice services cannot be restarted TEPHRA-179
+    Injector injector = Guice.createInjector(
+      new ConfigModule(conf),
+      new AbstractModule() {
+        @Override
+        protected void configure() {
+          // Instead of using ZKClientModule that will create new instance of ZKClient, we create instance
+          // binding to reuse the same ZKClient used for leader election
+          bind(ZKClient.class).toInstance(zkClientService);
+          bind(ZKClientService.class).toInstance(zkClientService);
+        }
+      },
+      new TransactionClientModule(),
+      getTransactionModule()
+    );
+    return injector.getInstance(TransactionManager.class);
+  }
+
+  @VisibleForTesting
+  protected Module getTransactionModule() {
+    return new TransactionModules().getDistributedModules();
+  }
+}