FAB-8842 Channel init retry

Change-Id: Idf00862d3761138a2d115f73f3128acf5ed7715a
Signed-off-by: rickr <cr22rc@gmail.com>

Author: cr22rc

pom.xml

@@ -189,7 +189,7 @@
         <dependency>
             <groupId>org.yaml</groupId>
             <artifactId>snakeyaml</artifactId>
-            <version>1.19</version>
+            <version>1.20</version>
         </dependency>
 
     </dependencies>

src/main/java/org/hyperledger/fabric/sdk/Channel.java

@@ -17,9 +17,9 @@
 import java.util.concurrent.TimeUnit;
 
 import com.google.common.util.concurrent.ListenableFuture;
+import io.grpc.ConnectivityState;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
-import io.grpc.StatusRuntimeException;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.hyperledger.fabric.protos.peer.EndorserGrpc;
@@ -86,24 +86,10 @@ public ListenableFuture<FabricProposalResponse.ProposalResponse> sendProposalAsy
         return futureStub.processProposal(proposal);
     }
 
-    public FabricProposalResponse.ProposalResponse sendProposal(FabricProposal.SignedProposal proposal) throws PeerException {
-
-        if (shutdown) {
-            throw new PeerException("Shutdown");
-        }
-
-        try {
-            return blockingStub.processProposal(proposal);
-
-        } catch (StatusRuntimeException e) {
-            logger.warn(String.format("RPC failed: %s", e.getStatus()));
-            throw new PeerException("Sending transaction to peer failed", e);
-        }
-    }
 
     boolean isChannelActive() {
         ManagedChannel lchannel = managedChannel;
-        return lchannel != null && !lchannel.isShutdown() && !lchannel.isTerminated();
+        return lchannel != null && !lchannel.isShutdown() && !lchannel.isTerminated() && ConnectivityState.READY.equals(lchannel.getState(true));
     }
 
     @Override

src/main/java/org/hyperledger/fabric/sdk/EndorserClient.java

@@ -74,6 +74,9 @@ public class EventHub implements Serializable {
     private Channel channel;
     private transient TransactionContext transactionContext;
     private transient byte[] clientTLSCertificateDigest;
+    private transient long reconnectCount;
+    private transient long lastBlockNumber;
+    private transient BlockEvent lastBlockEvent;
 
     /**
      * Get disconnected time.
@@ -167,19 +170,13 @@ public Properties getProperties() {
         return properties == null ? null : (Properties) properties.clone();
     }
 
-    boolean connect() throws EventHubException {
-
-        if (transactionContext == null) {
-            throw new EventHubException("Eventhub reconnect failed with no user context");
-        }
-
-        return connect(transactionContext);
-
-    }
-
     private transient StreamObserver<PeerEvents.Event> eventStream = null; // Saved here to avoid potential garbage collection
 
     synchronized boolean connect(final TransactionContext transactionContext) throws EventHubException {
+        return connect(transactionContext, false);
+    }
+
+    synchronized boolean connect(final TransactionContext transactionContext, final boolean reconnection) throws EventHubException {
         if (connected) {
             logger.warn(format("%s already connected.", toString()));
             return true;
@@ -209,72 +206,81 @@ public void onNext(PeerEvents.Event event) {
 
                 if (event.getEventCase() == PeerEvents.Event.EventCase.BLOCK) {
                     try {
-                        eventQue.addBEvent(new BlockEvent(EventHub.this, event));  //add to channel queue
+
+                        BlockEvent blockEvent = new BlockEvent(EventHub.this, event);
+                        setLastBlockSeen(blockEvent);
+
+                        eventQue.addBEvent(blockEvent);  //add to channel queue
                     } catch (InvalidProtocolBufferException e) {
                         EventHubException eventHubException = new EventHubException(format("%s onNext error %s", this, e.getMessage()), e);
                         logger.error(eventHubException.getMessage());
                         threw.add(eventHubException);
                     }
                 } else if (event.getEventCase() == PeerEvents.Event.EventCase.REGISTER) {
 
+                    if (reconnectCount > 1) {
+                        logger.info(format("Eventhub %s has reconnecting after %d attempts", name, reconnectCount));
+                    }
+
                     connected = true;
                     connectedTime = System.currentTimeMillis();
+                    reconnectCount = 0L;
+
                     finishLatch.countDown();
                 }
             }
 
             @Override
             public void onError(Throwable t) {
+                connected = false;
+                eventStream = null;
+                disconnectedTime = System.currentTimeMillis();
                 if (shutdown) { //IF we're shutdown don't try anything more.
                     logger.trace(format("%s was shutdown.", EventHub.this.toString()));
-                    connected = false;
-                    eventStream = null;
+
                     finishLatch.countDown();
                     return;
                 }
 
-                final boolean isTerminated = managedChannel.isTerminated();
-                final boolean isChannelShutdown = managedChannel.isShutdown();
+                final ManagedChannel lmanagedChannel = managedChannel;
+
+                final boolean isTerminated = lmanagedChannel == null ? true : lmanagedChannel.isTerminated();
+                final boolean isChannelShutdown = lmanagedChannel == null ? true : lmanagedChannel.isShutdown();
+
+                if (reconnectCount % 50 == 1) {
+                    logger.warn(format("%s terminated is %b shutdown is %b, retry count %d  has error %s.", EventHub.this.toString(), isTerminated, isChannelShutdown,
+                            reconnectCount, t.getMessage()));
+                } else {
+                    logger.trace(format("%s terminated is %b shutdown is %b, retry count %d  has error %s.", EventHub.this.toString(), isTerminated, isChannelShutdown,
+                            reconnectCount, t.getMessage()));
+                }
 
-                logger.error(format("%s terminated is %b shutdown is %b has error %s ", EventHub.this.toString(), isTerminated, isChannelShutdown,
-                        t.getMessage()), new EventHubException(t));
-                threw.add(t);
                 finishLatch.countDown();
 
                 //              logger.error("Error in stream: " + t.getMessage(), new EventHubException(t));
                 if (t instanceof StatusRuntimeException) {
                     StatusRuntimeException sre = (StatusRuntimeException) t;
                     Status sreStatus = sre.getStatus();
-                    logger.error(format("%s :StatusRuntimeException Status %s.  Description %s ", EventHub.this, sreStatus + "", sreStatus.getDescription()));
-                    if (sre.getStatus().getCode() == Status.Code.INTERNAL || sre.getStatus().getCode() == Status.Code.UNAVAILABLE) {
-
-                        connected = false;
-                        eventStream = null;
-                        disconnectedTime = System.currentTimeMillis();
-                        try {
-                            if (!isChannelShutdown) {
-                                managedChannel.shutdownNow();
-                            }
-                            if (null != disconnectedHandler) {
-                                try {
-                                    disconnectedHandler.disconnected(EventHub.this);
-                                } catch (Exception e) {
-                                    logger.warn(format("Eventhub %s  %s", EventHub.this.name, e.getMessage()), e);
-                                    eventQue.eventError(e);
-                                }
-                            }
-                        } catch (Exception e) {
-                            logger.warn(format("Eventhub %s Failed shutdown msg:  %s", EventHub.this.name, e.getMessage()), e);
-                        }
+                    if (reconnectCount % 50 == 1) {
+                        logger.warn(format("%s :StatusRuntimeException Status %s.  Description %s ", EventHub.this, sreStatus + "", sreStatus.getDescription()));
+                    } else {
+                        logger.trace(format("%s :StatusRuntimeException Status %s.  Description %s ", EventHub.this, sreStatus + "", sreStatus.getDescription()));
                     }
+
+                    try {
+                        reconnect();
+                    } catch (Exception e) {
+                        logger.warn(format("Eventhub %s Failed shutdown msg:  %s", EventHub.this.name, e.getMessage()));
+                    }
+
                 }
 
             }
 
             @Override
             public void onCompleted() {
 
-                logger.warn(format("Stream completed %s", EventHub.this.toString()));
+                logger.debug(format("Stream completed %s", EventHub.this.toString()));
                 finishLatch.countDown();
 
             }
@@ -288,27 +294,19 @@ public void onCompleted() {
         }
 
         try {
-            if (!finishLatch.await(EVENTHUB_CONNECTION_WAIT_TIME, TimeUnit.MILLISECONDS)) {
-                EventHubException evh = new EventHubException(format("EventHub %s failed to connect in %s ms.", name, EVENTHUB_CONNECTION_WAIT_TIME));
-                logger.debug(evh.getMessage(), evh);
 
-                throw evh;
-            }
-            logger.trace(format("Eventhub %s Done waiting for reply!", name));
+            //On reconnection don't wait here.
 
-        } catch (InterruptedException e) {
-            logger.error(e);
-        }
+            if (!reconnection && !finishLatch.await(EVENTHUB_CONNECTION_WAIT_TIME, TimeUnit.MILLISECONDS)) {
 
-        if (!threw.isEmpty()) {
-            eventStream = null;
-            connected = false;
-            Throwable t = threw.iterator().next();
+                logger.warn(format("EventHub %s failed to connect in %s ms.", name, EVENTHUB_CONNECTION_WAIT_TIME));
 
-            EventHubException evh = new EventHubException(t.getMessage(), t);
-            logger.error(format("EventHub %s Error in stream. error: " + t.getMessage(), toString()), evh);
-            throw evh;
+            } else {
+                logger.trace(format("Eventhub %s Done waiting for reply!", name));
+            }
 
+        } catch (InterruptedException e) {
+            logger.error(e);
         }
 
         logger.debug(format("Eventhub %s connect is done with connect status: %b ", name, connected));
@@ -321,6 +319,24 @@ public void onCompleted() {
 
     }
 
+    private void reconnect() throws EventHubException {
+
+        final ManagedChannel lmanagedChannel = managedChannel;
+
+        if (lmanagedChannel != null) {
+            managedChannel = null;
+            lmanagedChannel.shutdownNow();
+        }
+
+        EventHubDisconnected ldisconnectedHandler = disconnectedHandler;
+        if (!shutdown && null != ldisconnectedHandler) {
+            ++reconnectCount;
+            ldisconnectedHandler.disconnected(this);
+
+        }
+
+    }
+
     private void blockListen(TransactionContext transactionContext) throws CryptoException {
 
         this.transactionContext = transactionContext;
@@ -371,11 +387,17 @@ public String toString() {
 
     public void shutdown() {
         shutdown = true;
+        lastBlockEvent = null;
+        lastBlockNumber = 0;
         connected = false;
         disconnectedHandler = null;
         channel = null;
         eventStream = null;
-        managedChannel.shutdownNow();
+        final ManagedChannel lmanagedChannel = managedChannel;
+        managedChannel = null;
+        if (lmanagedChannel != null) {
+            lmanagedChannel.shutdownNow();
+        }
     }
 
     void setChannel(Channel channel) throws InvalidArgumentException {
@@ -391,6 +413,15 @@ void setChannel(Channel channel) throws InvalidArgumentException {
         this.channel = channel;
     }
 
+    synchronized void setLastBlockSeen(BlockEvent lastBlockSeen) {
+        long newLastBlockNumber = lastBlockSeen.getBlockNumber();
+        // overkill but make sure.
+        if (lastBlockNumber < newLastBlockNumber) {
+            lastBlockNumber = newLastBlockNumber;
+            this.lastBlockEvent = lastBlockSeen;
+        }
+    }
+
     /**
      * Eventhub disconnection notification interface
      */
@@ -412,32 +443,26 @@ public interface EventHubDisconnected {
 
     protected transient EventHubDisconnected disconnectedHandler = new EventHub.EventHubDisconnected() {
         @Override
-        public synchronized void disconnected(final EventHub eventHub) throws EventHubException {
-            logger.info(format("Detected disconnect %s", eventHub.toString()));
-
-            if (eventHub.connectedTime == 0) { //means event hub never connected
-                logger.error(format("%s failed on first connect no retries", eventHub.toString()));
-
-                eventHub.setEventHubDisconnectedHandler(null); //don't try again
-
-                //event hub never connected.
-                throw new EventHubException(format("%s never connected.", eventHub.toString()));
+        public synchronized void disconnected(final EventHub eventHub) {
+            if (reconnectCount == 1) {
+                logger.warn(format("Channel %s detected disconnect on event hub %s (%s)", channel.getName(), eventHub.toString(), url));
             }
 
             executorService.execute(() -> {
 
                 try {
                     Thread.sleep(500);
 
-                    if (eventHub.connect()) {
-                        logger.info(format("Successful reconnect %s", eventHub.toString()));
-                    } else {
-                        logger.info(format("Failed reconnect %s", eventHub.toString()));
+                    if (transactionContext == null) {
+                        logger.warn("Eventhub reconnect failed with no user context");
+                        return;
                     }
 
+                    eventHub.connect(transactionContext, true);
+
                 } catch (Exception e) {
 
-                    logger.debug(format("Failed %s to reconnect. %s", toString(), e.getMessage()));
+                    logger.warn(format("Failed %s to reconnect. %s", toString(), e.getMessage()));
 
                 }
 

src/main/java/org/hyperledger/fabric/sdk/EventHub.java

@@ -205,4 +205,9 @@ protected void finalize() throws Throwable {
         shutdown(true);
         super.finalize();
     }
+
+    @Override
+    public String toString() {
+        return "Orderer: " + name + "(" + url + ")";
+    }
 } // end Orderer

src/main/java/org/hyperledger/fabric/sdk/Orderer.java

@@ -20,6 +20,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
+import io.grpc.ConnectivityState;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.stub.StreamObserver;
@@ -326,6 +327,6 @@ public void onCompleted() {
 
     boolean isChannelActive() {
         ManagedChannel lchannel = managedChannel;
-        return lchannel != null && !lchannel.isShutdown() && !lchannel.isTerminated();
+        return lchannel != null && !lchannel.isShutdown() && !lchannel.isTerminated() && ConnectivityState.READY.equals(lchannel.getState(true));
     }
 }