|
1 | 1 | package com.alibaba.otter.canal.adapter.launcher.loader;
|
2 | 2 |
|
3 |
| -import java.util.ArrayList; |
4 |
| -import java.util.List; |
5 |
| -import java.util.Properties; |
6 |
| -import java.util.concurrent.ExecutorService; |
7 |
| -import java.util.concurrent.Future; |
8 |
| -import java.util.concurrent.TimeUnit; |
9 |
| -import java.util.concurrent.TimeoutException; |
10 |
| - |
11 |
| -import org.slf4j.Logger; |
12 |
| -import org.slf4j.LoggerFactory; |
13 |
| - |
14 | 3 | import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch;
|
15 | 4 | import com.alibaba.otter.canal.adapter.launcher.config.SpringContext;
|
16 | 5 | import com.alibaba.otter.canal.client.adapter.OuterAdapter;
|
|
23 | 12 | import com.alibaba.otter.canal.connector.core.spi.CanalMsgConsumer;
|
24 | 13 | import com.alibaba.otter.canal.connector.core.spi.ExtensionLoader;
|
25 | 14 | import com.alibaba.otter.canal.connector.core.spi.ProxyCanalMsgConsumer;
|
| 15 | +import org.slf4j.Logger; |
| 16 | +import org.slf4j.LoggerFactory; |
| 17 | + |
| 18 | +import java.util.ArrayList; |
| 19 | +import java.util.List; |
| 20 | +import java.util.Properties; |
| 21 | +import java.util.concurrent.ExecutorService; |
| 22 | +import java.util.concurrent.Future; |
| 23 | +import java.util.concurrent.TimeUnit; |
| 24 | +import java.util.concurrent.TimeoutException; |
26 | 25 |
|
27 | 26 | /**
|
28 | 27 | * 适配处理器
|
@@ -63,8 +62,9 @@ public AdapterProcessor(CanalClientConfig canalClientConfig, String destination,
|
63 | 62 |
|
64 | 63 | // load connector consumer
|
65 | 64 | ExtensionLoader<CanalMsgConsumer> loader = new ExtensionLoader<>(CanalMsgConsumer.class);
|
| 65 | + String key = destination + "_" + groupId; |
66 | 66 | canalMsgConsumer = new ProxyCanalMsgConsumer(loader
|
67 |
| - .getExtension(canalClientConfig.getMode().toLowerCase(), destination, CONNECTOR_SPI_DIR, |
| 67 | + .getExtension(canalClientConfig.getMode().toLowerCase(), key, CONNECTOR_SPI_DIR, |
68 | 68 | CONNECTOR_STANDBY_SPI_DIR));
|
69 | 69 |
|
70 | 70 | Properties properties = canalClientConfig.getConsumerProperties();
|
|
0 commit comments