Ready: TTB TS Support

Makefile

@@ -1,4 +1,6 @@
-.PHONY: compile test unit-test integration-test protogen
+.PHONY: compile test unit-test integration-test integration-test-timeseries protogen
+
+RIAK_PORT ?= 8087
 
 compile:
 	mvn clean compile
@@ -9,7 +11,13 @@ unit-test:
 	mvn test
 
 integration-test:
-	mvn -Pitest,default -Dcom.basho.riak.2i=true -Dcom.basho.riak.yokozuna=true -Dcom.basho.riak.buckettype=true -Dcom.basho.riak.crdt=true -Dcom.basho.riak.lifecycle=true verify
+	mvn -Pitest,default -Dcom.basho.riak.2i=true -Dcom.basho.riak.yokozuna=true -Dcom.basho.riak.buckettype=true -Dcom.basho.riak.crdt=true -Dcom.basho.riak.lifecycle=true -Dcom.basho.riak.pbcport=$(RIAK_PORT) verify
+
+integration-test-timeseries:
+	mvn -Pitest,default -Dcom.basho.riak.buckettype=true -Dcom.basho.riak.crdt=true -Dcom.basho.riak.lifecycle=true -Dcom.basho.riak.timeseries=true -Dcom.basho.riak.pbcport=$(RIAK_PORT) verify
+
+integration-test-security:
+	mvn -Pitest,default -Dcom.basho.riak.security=true -Dcom.basho.riak.security.clientcert=true -Dcom.basho.riak.pbcport=$(RIAK_PORT) test-compile failsafe:integration-test
 
 protogen:
 	mvn -Pprotobuf-generate generate-sources

pom.xml

@@ -431,5 +431,10 @@
             <artifactId>antlr-complete</artifactId>
             <version>3.5.2</version>
         </dependency>
+        <dependency>
+            <groupId>org.erlang.otp</groupId>
+            <artifactId>jinterface</artifactId>
+            <version>1.5.6</version>
+        </dependency>
     </dependencies>
 </project>

src/main/java/com/basho/riak/client/core/DefaultNodeManager.java

@@ -20,23 +20,24 @@
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
  *
- * The default {@link NodeManager} used by {@link RiakCluster} if none is 
+ * The default {@link NodeManager} used by {@link RiakCluster} if none is
  * specified.
- * 
- * This NodeManager round-robins through a list of {@link RiakNode}s and attempts 
- * to execute the operation passed to it. If a node reports that it is 
- * health checking it is removed from the list until it sends an update that it 
- * is again running. If the selected node cannot accept the operation because all 
- * connections are in use or it unable to make a new connection, the next node in 
- * the list is tried until either the operation is accepted or all nodes have 
- * been tried. If no nodes are able to accept the operation its setException() 
+ *
+ * This NodeManager round-robins through a list of {@link RiakNode}s and attempts
+ * to execute the operation passed to it. If a node reports that it is
+ * health checking it is removed from the list until it sends an update that it
+ * is again running. If the selected node cannot accept the operation because all
+ * connections are in use or it unable to make a new connection, the next node in
+ * the list is tried until either the operation is accepted or all nodes have
+ * been tried. If no nodes are able to accept the operation its setException()
  * method is called with a {@link NoNodesAvailableException}.
- * 
+ *
  * @author Brian Roach <roach at basho dot com>
  * @since 2.0
  */
@@ -47,7 +48,7 @@ public class DefaultNodeManager implements NodeManager, NodeStateListener
     private final AtomicInteger index = new AtomicInteger();
     private final Logger logger = LoggerFactory.getLogger(DefaultNodeManager.class);
     private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
-    
+
     @Override
     public void init(List<RiakNode> nodes)
     {
@@ -73,7 +74,7 @@ public boolean executeOnNode(FutureOperation operation, RiakNode previousNode)
             {
                 int startIndex = index.getAndIncrement();
                 int currentIndex = startIndex;
-                
+
                 do
                 {
                     if (healthy.get(Math.abs(currentIndex % healthy.size())).execute(operation))
@@ -89,15 +90,15 @@ else if (healthy.size() == 1)
             {
                 executed = healthy.get(0).execute(operation);
             }
-            
+
             return executed;
         }
         finally
         {
             lock.readLock().unlock();
         }
     }
-    
+
     @Override
     public void nodeStateChanged(RiakNode node, State state)
     {
@@ -110,7 +111,7 @@ public void nodeStateChanged(RiakNode node, State state)
                     if (unhealthy.remove(node))
                     {
                         healthy.add(node);
-                        logger.info("NodeManager moved node to healthy list; {}:{}", 
+                        logger.info("NodeManager moved node to healthy list; {}:{}",
                                     node.getRemoteAddress(), node.getPort());
                     }
                 }
@@ -126,7 +127,7 @@ public void nodeStateChanged(RiakNode node, State state)
                     if (healthy.remove(node))
                     {
                         unhealthy.add(node);
-                        logger.info("NodeManager moved node to unhealthy list; {}:{}", 
+                        logger.info("NodeManager moved node to unhealthy list; {}:{}",
                                     node.getRemoteAddress(), node.getPort());
                     }
                 }
@@ -174,7 +175,7 @@ public void addNode(RiakNode newNode)
         {
             lock.writeLock().unlock();
         }
-        
+
     }
 
     @Override
@@ -194,12 +195,12 @@ public boolean removeNode(RiakNode node)
         {
             lock.writeLock().unlock();
         }
-        
+
         if (removed)
         {
             node.removeStateListener(this);
             node.shutdown();
-            logger.info("NodeManager removed and shutdown node; {}:{}", 
+            logger.info("NodeManager removed and shutdown node; {}:{}",
                         node.getRemoteAddress(), node.getPort());
         }
         return removed;

src/main/java/com/basho/riak/client/core/FutureOperation.java

@@ -16,9 +16,6 @@
 package com.basho.riak.client.core;
 
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.LinkedList;
@@ -28,6 +25,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.locks.ReentrantLock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * @author Brian Roach <roach at basho dot com>
@@ -262,7 +261,7 @@ synchronized final void setException(Throwable t)
 
     public synchronized final Object channelMessage()
     {
-        Object message = createChannelMessage();
+        final Object message = createChannelMessage();
         state = State.WRITTEN;
         return message;
     }
@@ -309,17 +308,35 @@ public final T get() throws InterruptedException, ExecutionException
     {
         latch.await();
 
+        throwExceptionIfSet();
+
+        if (null == converted)
+        {
+            tryConvertResponse();
+        }
+
+        return converted;
+    }
+
+    private void throwExceptionIfSet() throws ExecutionException
+    {
         if (exception != null)
         {
             throw new ExecutionException(exception);
         }
-        else if (null == converted)
+    }
+
+    private void tryConvertResponse() throws ExecutionException
+    {
+        try
         {
             converted = convert(rawResponse);
-
         }
-
-        return converted;
+        catch(IllegalArgumentException ex)
+        {
+            exception = ex;
+            throwExceptionIfSet();
+        }
     }
 
     @Override
@@ -331,13 +348,12 @@ public final T get(long timeout, TimeUnit unit) throws InterruptedException, Exe
         {
             throw new TimeoutException();
         }
-        else if (exception != null)
-        {
-            throw new ExecutionException(exception);
-        }
-        else if (null == converted)
+
+        throwExceptionIfSet();
+
+        if (null == converted)
         {
-            converted = convert(rawResponse);
+            tryConvertResponse();
         }
 
         return converted;
@@ -373,6 +389,16 @@ public final void await(long timeout, TimeUnit unit) throws InterruptedException
         latch.await(timeout, unit);
     }
 
+    protected U checkAndGetSingleResponse(List<U> responses)
+    {
+        if (responses.size() > 1)
+        {
+            LoggerFactory.getLogger(this.getClass()).error("Received {} responses when only one was expected.",
+                                                           responses.size());
+        }
+
+        return responses.get(0);
+    }
 
     private void stateCheck(State... allowedStates)
     {