add auto reconnect to or and fix rotate bug

Author: andyqzb

databus2-relay/databus2-event-producer-or/src/main/java/com/linkedin/databus2/producers/ORListener.java

@@ -139,7 +139,7 @@ public interface TransactionProcessor
   /** Flag to indicate the begining of the txn is seen. Used to indicate  **/
   private boolean _isBeginTxnSeen = false;
 
-  private BlockingQueue<BinlogEventV4> binlogEventQueue = null;
+  private BlockingQueue<BinlogEventV4> _binlogEventQueue = null;
 
   public ORListener(String name,
                     int currentFileNumber,
@@ -159,7 +159,7 @@ public ORListener(String name,
     _tableUriToSrcNameMap = tableUriToSrcNameMap;
     _schemaRegistryService = schemaRegistryService;
     _currFileNum = currentFileNumber;
-    binlogEventQueue = new LinkedBlockingQueue<BinlogEventV4>(maxQueueSize);
+    _binlogEventQueue = new LinkedBlockingQueue<BinlogEventV4>(maxQueueSize);
   }
 
   @Override
@@ -168,14 +168,13 @@ public void onEvents(BinlogEventV4 event)
     boolean isPut = false;
     do {
       try {
-        isPut = binlogEventQueue.offer(event, 100, TimeUnit.MILLISECONDS);
+        isPut = _binlogEventQueue.offer(event, 100, TimeUnit.MILLISECONDS);
       } catch (InterruptedException e) {
         _log.error("failed to put binlog event to binlogEventQueue event: " + event, e);
       }
     } while (!isPut && !isShutdownRequested());
   }
 
-
   private void processTableMapEvent(TableMapEvent tme)
   {
     String newTableName = tme.getDatabaseName().toString().toLowerCase() + "." + tme.getTableName().toString().toLowerCase();
@@ -681,18 +680,30 @@ public static long scn(int logId, int offset)
   public void run() {
     List<BinlogEventV4> eventList = new ArrayList<BinlogEventV4>();
     BinlogEventV4 event;
-    boolean isShutdown = false;
-    while (!isShutdown)
+    while (!isShutdownRequested())
     {
-      eventList.clear();
-      int eventNumber = binlogEventQueue.drainTo(eventList);
-      for (int i = 0; i < eventNumber; i++)
+      if (isPauseRequested())
       {
-        if (isShutdownRequested())
+        LOG.info("Pause requested for ORListener. Pausing !!");
+        signalPause();
+        LOG.info("Pausing. Waiting for resume command");
+        try
+        {
+          awaitUnPauseRequest();
+        }
+        catch (InterruptedException e)
         {
-          isShutdown = true;
-          break;
+          _log.info("Interrupted !!");
         }
+        LOG.info("Resuming ORListener !!");
+        signalResumed();
+        LOG.info("ORListener resumed !!");
+      }
+
+      eventList.clear();
+      int eventNumber = _binlogEventQueue.drainTo(eventList);
+      for (int i = 0; i < eventNumber; i++)
+      {
 
         event = eventList.get(i);
         if (event == null)
@@ -716,6 +727,15 @@ public void run() {
               continue;
             }
           }
+          else if (event instanceof RotateEvent)
+          {
+            RotateEvent re = (RotateEvent) event;
+            String fileName = re.getBinlogFileName().toString();
+            _log.info("File Rotated : FileName :" + fileName + ", _binlogFilePrefix :" + _binlogFilePrefix);
+            String fileNumStr = fileName.substring(fileName.lastIndexOf(_binlogFilePrefix) + _binlogFilePrefix.length() + 1);
+            _currFileNum = Integer.parseInt(fileNumStr);
+            continue;
+          }
 
           if ( ! _isBeginTxnSeen )
           {
@@ -752,14 +772,6 @@ else if ("ROLLBACK".equalsIgnoreCase(sql))
               continue;
             }
           }
-          else if (event instanceof RotateEvent)
-          {
-            RotateEvent re = (RotateEvent)event;
-            String fileName = re.getBinlogFileName().toString();
-            _log.info("File Rotated : FileName :" + fileName + ", _binlogFilePrefix :" + _binlogFilePrefix);
-            String fileNumStr = fileName.substring(fileName.lastIndexOf(_binlogFilePrefix) + _binlogFilePrefix.length() + 1);
-            _currFileNum = Integer.parseInt(fileNumStr);
-          }
           else if (event instanceof XidEvent)
           {
             XidEvent xe = (XidEvent)event;
@@ -768,6 +780,12 @@ else if (event instanceof XidEvent)
             endXtion(xe);
             continue;
           }
+          else if (event instanceof FormatDescriptionEvent)
+          {
+            // we don't need process this event
+            _log.info("received FormatDescriptionEvent event");
+            continue;
+          }
           else if (event instanceof WriteRowsEvent)
           {
             WriteRowsEvent wre = (WriteRowsEvent)event;
@@ -803,18 +821,13 @@ else if (event instanceof TableMapEvent)
             TableMapEvent tme = (TableMapEvent)event;
             processTableMapEvent(tme);
           }
-          else if (event instanceof FormatDescriptionEvent)
-          {
-            // we don't need process this event
-            _log.info("receive FormatDescriptionEvent event");
-          }
           else
           {
             _log.warn("Skipping !! Unknown OR event e: " + event);
             continue;
           }
 
-          if ( _log.isDebugEnabled())
+          if (_log.isDebugEnabled())
           {
             _log.debug("e: " + event);
           }
@@ -824,5 +837,6 @@ else if (event instanceof FormatDescriptionEvent)
       }
     }
     _log.info("ORListener Thread done");
+    doShutdownNotify();
   }
 }

databus2-relay/databus2-event-producer-or/src/main/java/com/linkedin/databus2/producers/OpenReplicatorEventProducer.java

@@ -312,93 +312,171 @@ public class EventProducerThread extends DatabusThreadBase implements Transactio
 
     private String _sourceName;
 
+    private final long _reconnectIntervalMs = 2000;
+    private final long _workIntervalMs = 100;
+
     public EventProducerThread(String sourceName, long sinceScn)
     {
       super("OpenReplicator_" + sourceName);
       _sourceName = sourceName;
       _sinceScn = sinceScn;
     }
 
-    @Override
-    public void run()
+    void initOpenReplicator(long scn)
     {
-      _eventBuffer.start(_sinceScn);
-      _startPrevScn.set(_sinceScn);
-
-      int offset = offset(_sinceScn);
-      int logid = logid(_sinceScn);
+      int offset = offset(scn);
+      int logid = logid(scn);
 
       String binlogFile = String.format("%s.%06d", _binlogFilePrefix, logid);
-      _orListener = new ORListener(_sourceName, logid, _log,
-          _binlogFilePrefix, _producerThread, _tableUriToSrcIdMap,
+      // we should use a new ORListener to drop the left events in binlogEventQueue and the half processed transaction.
+      _orListener = new ORListener(_sourceName, logid, _log, _binlogFilePrefix, _producerThread, _tableUriToSrcIdMap,
           _tableUriToSrcNameMap, _schemaRegistryService, 200);
 
       _or.setBinlogFileName(binlogFile);
       _or.setBinlogPosition(offset);
       _or.setBinlogEventListener(_orListener);
 
+      //must set transport and binlogParser to null to drop the old connection environment in reinit case
+      _or.setTransport(null);
+      _or.setBinlogParser(null);
+
+      _log.info(String.format("Open Replicator starting from %s@%d", binlogFile, offset));
+    }
+
+    @Override
+    public void run()
+    {
+      _eventBuffer.start(_sinceScn);
+      _startPrevScn.set(_sinceScn);
+
+      initOpenReplicator(_sinceScn);
       try
       {
-        _log.info(String.format("Open Replicator starting from %s@%d", binlogFile, offset));
         _or.start();
         _orListener.start();
       } catch (Exception e)
       {
         throw new DatabusRuntimeException("failed to start open replicator: " + e.getMessage(), e);
       }
-      _log.info("Event Producer Thread done");
 
-    }
-
-    @Override
-    public void onEndTransaction(Transaction txn)
-        throws DatabusException
-    {
-      if (! isShutdownRequested())
+      long lastConnectMs = System.currentTimeMillis();
+      while (!isShutdownRequested())
       {
         if (isPauseRequested())
         {
           LOG.info("Pause requested for OpenReplicator. Pausing !!");
           signalPause();
           LOG.info("Pausing. Waiting for resume command");
-          try { awaitUnPauseRequest(); } catch (InterruptedException e) { _log.info("Interrupted !!"); }
+          try
+          {
+            if (_orListener.isAlive())
+            {
+              _orListener.pause();
+            }
+            awaitUnPauseRequest();
+          }
+          catch (InterruptedException e)
+          {
+            _log.info("Interrupted !!");
+          }
           LOG.info("Resuming OpenReplicator !!");
+          if (_orListener.isAlive())
+          {
+            try
+            {
+              _orListener.unpause();
+            }
+            catch (InterruptedException e)
+            {
+              _log.info("Interrupted !!");
+            }
+          }
           signalResumed();
           LOG.info("OpenReplicator resumed !!");
         }
 
-        try
+        if (!_or.isRunning() && System.currentTimeMillis() - lastConnectMs > _reconnectIntervalMs)
         {
-          addTxnToBuffer(txn);
-          _maxSCNReaderWriter.saveMaxScn(txn.getScn());
+          lastConnectMs = System.currentTimeMillis();
+          try
+          {
+            //should stop orListener first to get the final maxScn used for init open replicator.
+            if (_orListener.isAlive())
+            {
+              _orListener.shutdown();
+            }
+            long maxScn = _maxSCNReaderWriter.getMaxScn();
+            _startPrevScn.set(maxScn);
+            initOpenReplicator(maxScn);
+            _or.start();
+            _orListener.start();
+            _log.info("start Open Replicator successfully");
+          }
+          catch (Exception e)
+          {
+            _log.error("failed to start Open Replicator", e);
+            if (_or.isRunning())
+            {
+              try
+              {
+                _or.stop(10, TimeUnit.SECONDS);
+              }
+              catch (Exception e2)
+              {
+                _log.error("failed to stop Open Replicator", e2);
+              }
+            }
+          }
         }
-        catch (UnsupportedKeyException e)
+
+        try
         {
-          _log.fatal("Got UnsupportedKeyException exception while adding txn (" + txn + ") to the buffer", e);
-          throw new DatabusException(e);
+          Thread.sleep(_workIntervalMs);
         }
-        catch (EventCreationException e)
+        catch (InterruptedException e)
         {
-          _log.fatal("Got EventCreationException exception while adding txn (" + txn + ") to the buffer", e);
-          throw new DatabusException(e);
+          _log.info("Interrupted !!");
         }
       }
 
-      if (isShutdownRequested())
+      if (_or.isRunning())
       {
         try
         {
-          // Because the current thread is orListener thread, so shutdown() will block the thread itself.
-          // So we just set orListener's shutdown flag, the orListener thead will exit immediately after this function.
-          _orListener.shutdownAsynchronously();
           _or.stop(10, TimeUnit.SECONDS);
         }
         catch (Exception e)
         {
-          _log.error("Got exception while stopping open replicator",e);
+          _log.error("failed to stop Open Replicator", e);
         }
-        doShutdownNotify();
-        return;
+      }
+      if (_orListener.isAlive())
+      {
+        _orListener.shutdown();
+      }
+
+      _log.info("Event Producer Thread done");
+      doShutdownNotify();
+    }
+
+    @Override
+    public void onEndTransaction(Transaction txn)
+        throws DatabusException
+    {
+      try
+      {
+        addTxnToBuffer(txn);
+        _maxSCNReaderWriter.saveMaxScn(txn.getScn());
+      }
+      catch (UnsupportedKeyException e)
+      {
+        _log.fatal("Got UnsupportedKeyException exception while adding txn (" + txn + ") to the buffer", e);
+        throw new DatabusException(e);
+      }
+      catch (EventCreationException e)
+      {
+        _log.fatal("Got EventCreationException exception while adding txn (" + txn + ") to the buffer", e);
+        throw new DatabusException(e);
       }
     }