refactor: Use Java standard library instead of Guava (#3708)

Co-authored-by: Moderne <team@moderne.io>

Co-authored-by: Moderne <team@moderne.io>

client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaOffsetCanalConnector.java

@@ -6,7 +6,6 @@ import com.alibaba.otter.canal.client.kafka.protocol.KafkaMessage;
 import com.alibaba.otter.canal.protocol.FlatMessage;
 import com.alibaba.otter.canal.protocol.Message;
 import com.alibaba.otter.canal.protocol.exception.CanalClientException;
-import com.google.common.collect.Lists;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -43,7 +42,7 @@ public class KafkaOffsetCanalConnector extends KafkaCanalConnector {
     public List<KafkaMessage> getListWithoutAck(Long timeout, TimeUnit unit, long offset) throws CanalClientException {
         waitClientRunning();
         if (!running) {
-            return Lists.newArrayList();
+            return new ArrayList<>();
         }
 
         if (offset > -1) {
@@ -61,7 +60,7 @@ public class KafkaOffsetCanalConnector extends KafkaCanalConnector {
             }
             return messages;
         }
-        return Lists.newArrayList();
+        return new ArrayList<>();
     }
 
     /**
@@ -76,7 +75,7 @@ public class KafkaOffsetCanalConnector extends KafkaCanalConnector {
     public List<KafkaFlatMessage> getFlatListWithoutAck(Long timeout, TimeUnit unit, long offset) throws CanalClientException {
         waitClientRunning();
         if (!running) {
-            return Lists.newArrayList();
+            return new ArrayList<>();
         }
 
         if (offset > -1) {
@@ -96,7 +95,7 @@ public class KafkaOffsetCanalConnector extends KafkaCanalConnector {
 
             return flatMessages;
         }
-        return Lists.newArrayList();
+        return new ArrayList<>();
     }
 
     /**

client/src/main/java/com/alibaba/otter/canal/client/rabbitmq/RabbitMQCanalConnector.java

@@ -13,6 +13,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -239,7 +240,7 @@ public class RabbitMQCanalConnector implements CanalMQConnector {
         if (logger.isDebugEnabled()) {
             logger.debug("Get Message: {}", new String(messageData));
         }
-        List messageList = Lists.newArrayList();
+        List messageList = new ArrayList<>();
         if (!flatMessage) {
             Message message = CanalMessageDeserializer.deserializer(messageData);
             messageList.add(message);

client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java

@@ -1,5 +1,6 @@
 package com.alibaba.otter.canal.client.rocketmq;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -165,7 +166,7 @@ public class RocketMQCanalConnector implements CanalMQConnector {
         if (logger.isDebugEnabled()) {
             logger.debug("Get Message: {}", messageExts);
         }
-        List messageList = Lists.newArrayList();
+        List messageList = new ArrayList<>();
         for (MessageExt messageExt : messageExts) {
             byte[] data = messageExt.getBody();
             if (data != null) {

connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java

@@ -36,7 +36,7 @@ public class MQMessageUtils {
     private static Map<String, List<PartitionData>>    partitionDatas    = MigrateMap.makeComputingMap(CacheBuilder.newBuilder()
                                                                              .softValues(),
                                                                              pkHashConfigs -> {
-                                                                                 List<PartitionData> datas = Lists.newArrayList();
+                                                                                 List<PartitionData> datas = new ArrayList<>();
 
                                                                                  String[] pkHashConfigArray = StringUtils.split(StringUtils.replace(pkHashConfigs,
                                                                                      ",",
@@ -75,7 +75,7 @@ public class MQMessageUtils {
     private static Map<String, List<DynamicTopicData>> dynamicTopicDatas = MigrateMap.makeComputingMap(CacheBuilder.newBuilder()
                                                                              .softValues(),
                                                                               pkHashConfigs -> {
-                                                                                List<DynamicTopicData> datas = Lists.newArrayList();
+                                                                                List<DynamicTopicData> datas = new ArrayList<>();
                                                                                 String[] dynamicTopicArray = StringUtils.split(StringUtils.replace(pkHashConfigs,
                                                                                     ",",
                                                                                     ";"),
@@ -102,7 +102,7 @@ public class MQMessageUtils {
     private static Map<String, List<TopicPartitionData>> topicPartitionDatas = MigrateMap.makeComputingMap(CacheBuilder.newBuilder()
                                                                                 .softValues(),
                                                                                 tPConfigs -> {
-                                                                                    List<TopicPartitionData> datas = Lists.newArrayList();
+                                                                                    List<TopicPartitionData> datas = new ArrayList<>();
                                                                                     String[] tPArray = StringUtils.split(StringUtils.replace(tPConfigs,
                                                                                             ",",
                                                                                             ";"),
@@ -258,7 +258,7 @@ public class MQMessageUtils {
         List<Entry>[] partitionEntries = new List[partitionsNum];
         for (int i = 0; i < partitionsNum; i++) {
             // 注意一下并发
-            partitionEntries[i] = Collections.synchronizedList(Lists.newArrayList());
+            partitionEntries[i] = Collections.synchronizedList(new ArrayList<>());
         }
 
         for (EntryRowData data : datas) {
@@ -681,7 +681,7 @@ public class MQMessageUtils {
 
         public boolean      autoPkHash = false;
         public boolean      tableHash  = false;
-        public List<String> pkNames    = Lists.newArrayList();
+        public List<String> pkNames    = new ArrayList<>();
     }
 
     public static class DynamicTopicData {

connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/consumer/CanalRabbitMQConsumer.java

@@ -1,6 +1,7 @@
 package com.alibaba.otter.canal.connector.rabbitmq.consumer;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Properties;
 import java.util.concurrent.BlockingQueue;
@@ -23,7 +24,6 @@ import com.alibaba.otter.canal.connector.rabbitmq.config.RabbitMQConstants;
 import com.alibaba.otter.canal.connector.rabbitmq.producer.AliyunCredentialsProvider;
 import com.alibaba.otter.canal.protocol.Message;
 import com.alibaba.otter.canal.protocol.exception.CanalClientException;
-import com.google.common.collect.Lists;
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
@@ -137,7 +137,7 @@ public class CanalRabbitMQConsumer implements CanalMsgConsumer {
         if (logger.isDebugEnabled()) {
             logger.debug("Get Message: {}", new String(messageData));
         }
-        List<CommonMessage> messageList = Lists.newArrayList();
+        List<CommonMessage> messageList = new ArrayList<>();
         if (!flatMessage) {
             Message message = CanalMessageSerializerUtil.deserializer(messageData);
             messageList.addAll(MessageUtil.convert(message));

connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/consumer/CanalRocketMQConsumer.java

@@ -1,5 +1,6 @@
 package com.alibaba.otter.canal.connector.rocketmq.consumer;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Properties;
 import java.util.concurrent.BlockingQueue;
@@ -30,7 +31,6 @@ import com.alibaba.otter.canal.connector.core.util.MessageUtil;
 import com.alibaba.otter.canal.connector.rocketmq.config.RocketMQConstants;
 import com.alibaba.otter.canal.protocol.Message;
 import com.alibaba.otter.canal.protocol.exception.CanalClientException;
-import com.google.common.collect.Lists;
 
 /**
  * RocketMQ consumer SPI 实现
@@ -142,7 +142,7 @@ public class CanalRocketMQConsumer implements CanalMsgConsumer {
         if (logger.isDebugEnabled()) {
             logger.debug("Get Message: {}", messageExts);
         }
-        List<CommonMessage> messageList = Lists.newArrayList();
+        List<CommonMessage> messageList = new ArrayList<>();
         for (MessageExt messageExt : messageExts) {
             byte[] data = messageExt.getBody();
             if (data != null) {

deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerInstanceConfigMonitor.java

@@ -1,5 +1,6 @@
 package com.alibaba.otter.canal.deployer.monitor;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Executors;
@@ -77,9 +78,9 @@ public class ManagerInstanceConfigMonitor extends AbstractCanalLifeCycle impleme
         }
 
         final List<String> is = Lists.newArrayList(StringUtils.split(instances, ','));
-        List<String> start = Lists.newArrayList();
-        List<String> stop = Lists.newArrayList();
-        List<String> restart = Lists.newArrayList();
+        List<String> start = new ArrayList<>();
+        List<String> stop = new ArrayList<>();
+        List<String> restart = new ArrayList<>();
         for (String instance : is) {
             if (!configs.containsKey(instance)) {
                 PlainCanal newPlainCanal = configClient.findInstance(instance, null);

meta/src/main/java/com/alibaba/otter/canal/meta/FileMixedMetaManager.java

@@ -194,7 +194,7 @@ public class FileMixedMetaManager extends MemoryMetaManager implements CanalMeta
             synchronized (destination.intern()) { // 基于destination控制一下并发更新
                 data.setDestination(destination);
 
-                List<FileMetaClientIdentityData> clientDatas = Lists.newArrayList();
+                List<FileMetaClientIdentityData> clientDatas = new ArrayList<>();
                 List<ClientIdentity> clientIdentitys = destinations.get(destination);
                 for (ClientIdentity clientIdentity : clientIdentitys) {
                     FileMetaClientIdentityData clientData = new FileMetaClientIdentityData();

meta/src/main/java/com/alibaba/otter/canal/meta/MemoryMetaManager.java

@@ -1,9 +1,6 @@
 package com.alibaba.otter.canal.meta;
 
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.atomic.AtomicLong;
 
 import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
@@ -35,7 +32,7 @@ public class MemoryMetaManager extends AbstractCanalLifeCycle implements CanalMe
 
         cursors = new MapMaker().makeMap();
 
-        destinations = MigrateMap.makeComputingMap(destination -> Lists.newArrayList());
+        destinations = MigrateMap.makeComputingMap(destination -> new ArrayList<>());
     }
 
     public void stop() {
@@ -186,8 +183,8 @@ public class MemoryMetaManager extends AbstractCanalLifeCycle implements CanalMe
 
         public synchronized Map<Long, PositionRange> listAllPositionRange() {
             Set<Long> batchIdSets = batches.keySet();
-            List<Long> batchIds = Lists.newArrayList(batchIdSets);
-            Collections.sort(Lists.newArrayList(batchIds));
+            List<Long> batchIds = new ArrayList<>(batchIdSets);
+            Collections.sort(new ArrayList<>(batchIds));
 
             return Maps.newHashMap(batches);
         }

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta_DDL_Test.java

@@ -3,6 +3,7 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb;
 import java.io.File;
 import java.io.FileInputStream;
 import java.net.URL;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.commons.io.IOUtils;
@@ -15,7 +16,6 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 import com.alibaba.druid.sql.repository.Schema;
 import com.alibaba.otter.canal.parse.inbound.TableMeta;
-import com.google.common.collect.Lists;
 
 /**
  * @author agapple 2017年8月1日 下午7:15:54
@@ -79,7 +79,7 @@ public class MemoryTableMeta_DDL_Test {
         String sql = StringUtils.join(IOUtils.readLines(new FileInputStream(create)), "\n");
         memoryTableMeta.apply(null, "test", sql, null);
 
-        List<String> tableNames = Lists.newArrayList();
+        List<String> tableNames = new ArrayList<>();
         for (Schema schema : memoryTableMeta.getRepository().getSchemas()) {
             tableNames.addAll(schema.showTables());
         }
@@ -99,7 +99,7 @@ public class MemoryTableMeta_DDL_Test {
         String sql = StringUtils.join(IOUtils.readLines(new FileInputStream(create)), "\n");
         memoryTableMeta.apply(null, "test", sql, null);
 
-        List<String> tableNames = Lists.newArrayList();
+        List<String> tableNames = new ArrayList<>();
         for (Schema schema : memoryTableMeta.getRepository().getSchemas()) {
             tableNames.addAll(schema.showTables());
         }

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta_Random_DDL_Test.java

@@ -3,6 +3,7 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb;
 import java.io.File;
 import java.io.FileInputStream;
 import java.net.URL;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.commons.io.IOUtils;
@@ -15,7 +16,6 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 import com.alibaba.druid.sql.repository.Schema;
 import com.alibaba.otter.canal.parse.inbound.TableMeta;
-import com.google.common.collect.Lists;
 
 /**
  * @author agapple 2017年8月1日 下午7:15:54
@@ -69,7 +69,7 @@ public class MemoryTableMeta_Random_DDL_Test {
     }
 
     private void compareTableMeta(int num, MemoryTableMeta source, MemoryTableMeta target) {
-        List<String> tableNames = Lists.newArrayList();
+        List<String> tableNames = new ArrayList<>();
         for (Schema schema : source.getRepository().getSchemas()) {
             tableNames.addAll(schema.showTables());
         }

protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java

@@ -1,11 +1,10 @@
 package com.alibaba.otter.canal.protocol;
 
 import java.io.Serializable;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
-import com.google.common.collect.Lists;
-
 /**
  * @author machengyuan 2018-9-13 下午10:31:14
  * @version 1.0.0
@@ -66,7 +65,7 @@ public class FlatMessage implements Serializable {
 
     public void addPkName(String pkName) {
         if (this.pkNames == null) {
-            this.pkNames = Lists.newArrayList();
+            this.pkNames = new ArrayList<>();
         }
         this.pkNames.add(pkName);
     }