Merge pull request #293 from MarkLinHz/master

fix bug of repeat log manager

deployer/src/main/resources/logback.xml

@@ -74,10 +74,8 @@
         <appender-ref ref="CANAL-META" />
     </logger>
     
-	<root level="WARN">
-		<!--  
-		<appender-ref ref="STDOUT"/>
-		-->
+	<root level="INFO">
+		<!--<appender-ref ref="STDOUT"/>-->
 		<appender-ref ref="CANAL-ROOT" />
 	</root>
 </configuration>

parse/src/main/java/com/alibaba/otter/canal/parse/index/FailbackLogPositionManager.java

@@ -42,7 +42,7 @@ public class FailbackLogPositionManager extends AbstractLogPositionManager {
         }
 
         if (!secondary.isStart()) {
-            primary.start();
+            secondary.start();
         }
     }
 

parse/src/main/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManager.java

@@ -38,8 +38,13 @@ public class MixedLogPositionManager extends AbstractLogPositionManager {
     public void start() {
         super.start();
 
-        memoryLogPositionManager.start();
-        zooKeeperLogPositionManager.start();
+        if (!memoryLogPositionManager.isStart()) {
+            memoryLogPositionManager.start();
+        }
+
+        if (!zooKeeperLogPositionManager.isStart()) {
+            zooKeeperLogPositionManager.start();
+        }
     }
 
     @Override

parse/src/test/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManagerTest.java

@@ -45,8 +45,7 @@ public class MetaLogPositionManagerTest extends AbstractLogPositionManagerTest {
         metaManager.setZooKeeperMetaManager(zooKeeperMetaManager);
         metaManager.start();
 
-        MetaLogPositionManager logPositionManager = new MetaLogPositionManager();
-        logPositionManager.setMetaManager(metaManager);
+        MetaLogPositionManager logPositionManager = new MetaLogPositionManager(metaManager);
         logPositionManager.start();
         // 构建meta信息
         ClientIdentity client1 = new ClientIdentity(destination, (short) 1);

parse/src/test/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManagerTest.java

@@ -27,19 +27,16 @@ public class MixedLogPositionManagerTest extends AbstractLogPositionManagerTest
 
     @Test
     public void testAll() {
-        MixedLogPositionManager logPositionManager = new MixedLogPositionManager();
+        MemoryLogPositionManager memoryLogPositionManager = new MemoryLogPositionManager();
+        ZooKeeperLogPositionManager zookeeperLogPositionManager = new ZooKeeperLogPositionManager(zkclientx);
 
-        ZooKeeperLogPositionManager zookeeperLogPositionManager = new ZooKeeperLogPositionManager();
-        zookeeperLogPositionManager.setZkClientx(zkclientx);
-
-        logPositionManager.setZooKeeperLogPositionManager(zookeeperLogPositionManager);
+        MixedLogPositionManager logPositionManager = new MixedLogPositionManager(zkclientx);
         logPositionManager.start();
 
         LogPosition position2 = doTest(logPositionManager);
         sleep(1000);
 
-        MixedLogPositionManager logPositionManager2 = new MixedLogPositionManager();
-        logPositionManager2.setZooKeeperLogPositionManager(zookeeperLogPositionManager);
+        MixedLogPositionManager logPositionManager2 = new MixedLogPositionManager(zkclientx);
         logPositionManager2.start();
 
         LogPosition getPosition2 = logPositionManager2.getLatestIndexBy(destination);

parse/src/test/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManagerTest.java

@@ -27,19 +27,17 @@ public class PeriodMixedLogPositionManagerTest extends AbstractLogPositionManage
 
     @Test
     public void testAll() {
-        PeriodMixedLogPositionManager logPositionManager = new PeriodMixedLogPositionManager();
+        MemoryLogPositionManager memoryLogPositionManager = new MemoryLogPositionManager();
+        ZooKeeperLogPositionManager zookeeperLogPositionManager = new ZooKeeperLogPositionManager(zkclientx);
 
-        ZooKeeperLogPositionManager zookeeperLogPositionManager = new ZooKeeperLogPositionManager();
-        zookeeperLogPositionManager.setZkClientx(zkclientx);
+        PeriodMixedLogPositionManager logPositionManager = new PeriodMixedLogPositionManager(memoryLogPositionManager, zookeeperLogPositionManager, 1000L);
 
-        logPositionManager.setZooKeeperLogPositionManager(zookeeperLogPositionManager);
         logPositionManager.start();
 
         LogPosition position2 = doTest(logPositionManager);
         sleep(1500);
 
-        PeriodMixedLogPositionManager logPositionManager2 = new PeriodMixedLogPositionManager();
-        logPositionManager2.setZooKeeperLogPositionManager(zookeeperLogPositionManager);
+        PeriodMixedLogPositionManager logPositionManager2 = new PeriodMixedLogPositionManager(memoryLogPositionManager, zookeeperLogPositionManager, 1000L);
         logPositionManager2.start();
 
         LogPosition getPosition2 = logPositionManager2.getLatestIndexBy(destination);

parse/src/test/java/com/alibaba/otter/canal/parse/index/ZooKeeperLogPositionManagerTest.java

@@ -25,8 +25,7 @@ public class ZooKeeperLogPositionManagerTest extends AbstractLogPositionManagerT
 
     @Test
     public void testAll() {
-        ZooKeeperLogPositionManager logPositionManager = new ZooKeeperLogPositionManager();
-        logPositionManager.setZkClientx(zkclientx);
+        ZooKeeperLogPositionManager logPositionManager = new ZooKeeperLogPositionManager(zkclientx);
         logPositionManager.start();
 
         doTest(logPositionManager);