diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index 548df70c..e4f437f8 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -13,6 +13,7 @@ import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.springframework.util.CollectionUtils; +import com.alibaba.otter.canal.common.utils.JsonUtils; import com.alibaba.otter.canal.parse.CanalEventParser; import com.alibaba.otter.canal.parse.CanalHASwitchable; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.FieldPacket; @@ -410,7 +411,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE } } // 其余情况 - logger.warn("prepare to find start position just last position"); + logger.warn("prepare to find start position just last position\n {}",JsonUtils.marshalToString(logPosition)); return logPosition.getPostion(); } else { // 针对切换的情况,考虑回退时间 diff --git a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java index d8488668..7123e389 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java @@ -131,6 +131,8 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C */ @Override public void subscribe(ClientIdentity clientIdentity) throws CanalServerException { + checkStart(clientIdentity.getDestination()); + CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination()); if (!canalInstance.getMetaManager().isStart()) { canalInstance.getMetaManager().start(); diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java index ce0a0790..4cefb90c 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java @@ -66,8 +66,7 @@ public class SessionHandler extends SimpleChannelHandler { Short.valueOf(sub.getClientId()), sub.getFilter()); MDC.put("destination", clientIdentity.getDestination()); - embeddedServer.subscribe(clientIdentity); - + // 尝试启动,如果已经启动,忽略 if (!embeddedServer.isStart(clientIdentity.getDestination())) { ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination()); @@ -75,6 +74,8 @@ public class SessionHandler extends SimpleChannelHandler { runningMonitor.start(); } } + + embeddedServer.subscribe(clientIdentity); ctx.setAttachment(clientIdentity);// 设置状态数据 NettyUtils.ack(ctx.getChannel(), null);