Cleanup all threads on #close()

hbase-commit-table/src/main/java/com/yahoo/omid/committable/hbase/HBaseCommitTable.java

@@ -215,6 +215,10 @@ public void run() {
         public void close() throws IOException {
             deleteBatchExecutor.shutdown();
             try {
+                if (!deleteBatchExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
+                    deleteBatchExecutor.shutdownNow(); // may need to interrupt take
+                }
+
                 if (!deleteBatchExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
                     LOG.warn("Delete executor did not shutdown");
                 }

tso-client/src/main/java/com/yahoo/omid/client/TSOClient.java

@@ -82,6 +82,7 @@ public class TSOClient {
 
     private ChannelFactory factory;
     private ClientBootstrap bootstrap;
+    private final ScheduledExecutorService fsmExecutor;
     Fsm fsm;
 
     private final int requestTimeoutMs;
@@ -165,7 +166,7 @@ private TSOClient(Configuration conf, MetricRegistry metrics) {
 
         addr = new InetSocketAddress(host, port);
 
-        ScheduledExecutorService fsmExecutor = Executors.newSingleThreadScheduledExecutor(
+        fsmExecutor = Executors.newSingleThreadScheduledExecutor(
                 new ThreadFactoryBuilder().setNameFormat("tsofsm-%d").build());
         fsm = new FsmImpl(fsmExecutor);
         fsm.setInitState(new DisconnectedState());
@@ -214,6 +215,12 @@ public TSOFuture<Long> commit(long transactionId, Set<? extends CellId> cells) {
     public TSOFuture<Void> close() {
         CloseEvent closeEvent = new CloseEvent();
         fsm.sendEvent(closeEvent);
+        closeEvent.addListener(new Runnable() {
+                @Override
+                public void run() {
+                    fsmExecutor.shutdown();
+                }
+            }, fsmExecutor);
         return new ForwardingTSOFuture<Void>(closeEvent);
     }
 
@@ -323,6 +330,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
                     });
                 return new ConnectingState(retries - 1);
             } else if (e instanceof CloseEvent) {
+                factory.releaseExternalResources();
                 ((CloseEvent)e).success(null);
                 return this;
             } else {