|
|
@@ -13,7 +13,6 @@ import org.slf4j.LoggerFactory;
|
|
|
import com.alibaba.otter.canal.common.MQProperties;
|
|
|
import com.alibaba.otter.canal.instance.core.CanalInstance;
|
|
|
import com.alibaba.otter.canal.instance.core.CanalMQConfig;
|
|
|
-import com.alibaba.otter.canal.kafka.CanalKafkaProducer;
|
|
|
import com.alibaba.otter.canal.protocol.ClientIdentity;
|
|
|
import com.alibaba.otter.canal.protocol.Message;
|
|
|
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
|
|
|
@@ -146,7 +145,7 @@ public class CanalMQStarter {
|
|
|
try {
|
|
|
int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size();
|
|
|
if (batchId != -1 && size != 0) {
|
|
|
- canalMQProducer.send(destination, message, new CanalKafkaProducer.Callback() {
|
|
|
+ canalMQProducer.send(destination, message, new CanalMQProducer.Callback() {
|
|
|
|
|
|
@Override
|
|
|
public void commit() {
|
guixincui