chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fastjson update 1.x to 2.0.2 (#4160)

Browse Source
This commit is contained in:
温绍锦
2022-05-13 11:32:47 +08:00
committed by GitHub
parent bac016f89e
commit ead91433d7
35 changed files with 105 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
admin/admin-web/src/main/java/com/alibaba/otter/canal/admin/controller/CanalClusterController.java
View File

@@ -13,8 +13,8 @@ import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.otter.canal.admin.model.BaseModel;
import com.alibaba.otter.canal.admin.model.CanalCluster;
import com.alibaba.otter.canal.admin.model.NodeServer;

6
client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/service/ESSyncService.java
View File

@@ -8,11 +8,11 @@ import java.util.Map;
import javax.sql.DataSource;
import com.alibaba.fastjson2.JSONWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlSelectQueryBlock;
import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig;
import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig.ESMapping;
@@ -79,7 +79,7 @@ public class ESSyncService {
esSyncConfigs
.forEach(esSyncConfig -> configIndexes.append(esSyncConfig.getEsMapping().get_index()).append(" "));
logger.debug("DML: {} \nAffected indexes: {}",
JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue),
JSON.toJSONString(dml, JSONWriter.Feature.WriteNulls),
configIndexes.toString());
}
}

2
client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESSyncUtil.java
View File

@@ -18,7 +18,7 @@ import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig.ESMapping;
import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem;
import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.ColumnItem;

6
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java
View File

@@ -6,8 +6,8 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter.Feature;
import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.hbase.support.*;
import com.alibaba.otter.canal.client.adapter.support.Dml;
@@ -39,7 +39,7 @@ public class HbaseSyncService {
delete(config, dml);
}
if (logger.isDebugEnabled()) {
logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
logger.debug("DML: {}", JSON.toJSONString(dml, Feature.WriteNulls));
}
}
}

12
client-adapter/kudu/src/main/java/com/alibaba/otter/canal/client/adapter/kudu/service/KuduSyncService.java
View File

@@ -10,8 +10,8 @@ import org.apache.kudu.client.KuduException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter.Feature;
import com.alibaba.otter.canal.client.adapter.kudu.config.KuduMappingConfig;
import com.alibaba.otter.canal.client.adapter.kudu.support.KuduTemplate;
import com.alibaba.otter.canal.client.adapter.support.Dml;
@@ -55,7 +55,7 @@ public class KuduSyncService {
delete(config, dml);
}
if (logger.isDebugEnabled()) {
logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
logger.debug("DML: {}", JSON.toJSONString(dml, Feature.WriteNulls));
}
}
}
@@ -118,7 +118,7 @@ public class KuduSyncService {
}
} catch (KuduException e) {
logger.error(e.getMessage());
logger.error("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
logger.error("DML: {}", JSON.toJSONString(dml, Feature.WriteNulls));
}
}
}
@@ -159,7 +159,7 @@ public class KuduSyncService {
}
} catch (KuduException e) {
logger.error(e.getMessage());
logger.error("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
logger.error("DML: {}", JSON.toJSONString(dml, Feature.WriteNulls));
}
}
@@ -201,7 +201,7 @@ public class KuduSyncService {
}
} catch (KuduException e) {
logger.error(e.getMessage());
logger.error("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
logger.error("DML: {}", JSON.toJSONString(dml, Feature.WriteNulls));
}
}
}

6
client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java
View File

@@ -6,8 +6,8 @@ import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter.Feature;
import com.alibaba.otter.canal.client.adapter.OuterAdapter;
import com.alibaba.otter.canal.client.adapter.support.Dml;
import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
@@ -36,7 +36,7 @@ public class LoggerAdapterExample implements OuterAdapter {
}
public void sync(Dml dml) {
logger.info("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
logger.info("DML: {}", JSON.toJSONString(dml, Feature.WriteNulls));
}
@Override

6
client-adapter/phoenix/pom.xml
View File

@@ -75,9 +75,9 @@
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.58</version>
<groupId>com.alibaba.fastjson2</groupId>
<artifactId>fastjson2</artifactId>
<version>2.0.2</version>
<scope>provided</scope>
</dependency>

10
client-adapter/phoenix/src/main/java/com/alibaba/otter/canal/client/adapter/phoenix/service/PhoenixSyncService.java
View File

@@ -6,8 +6,8 @@ import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.statement.*;
import com.alibaba.druid.sql.parser.ParserException;
import com.alibaba.druid.util.JdbcConstants;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter.Feature;
import com.alibaba.otter.canal.client.adapter.phoenix.config.ConfigurationManager;
import com.alibaba.otter.canal.client.adapter.phoenix.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.phoenix.config.MappingConfig.DbMapping;
@@ -208,7 +208,7 @@ public class PhoenixSyncService {
for (Map.Entry<String, MappingConfig> entry : configMap.entrySet()) {
MappingConfig config = entry.getValue();
if (config.isDebug()) {
logger.info("DML: {} {}", entry.getKey(), JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
logger.info("DML: {} {}", entry.getKey(), JSON.toJSONString(dml, Feature.WriteNulls));
}
if (config.getConcurrent()) {
//并行同步
@@ -257,7 +257,7 @@ public class PhoenixSyncService {
} else if (type != null && type.equalsIgnoreCase("TRUNCATE")) {
truncate(batchExecutor, config);
} else if (logger.isInfoEnabled()){
logger.info("SingleDml: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
logger.info("SingleDml: {}", JSON.toJSONString(dml, Feature.WriteNulls));
}
} catch (SQLException e) {
logger.error("sync error: " + e.getMessage(), e);
@@ -268,7 +268,7 @@ public class PhoenixSyncService {
private void alter(BatchExecutor batchExecutor, MappingConfig config, Dml dml, List<SQLStatement> stmtList, String configFile) throws SQLException {
if (config.isDebug()) {
logger.info("DML: {} {}", configFile, JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
logger.info("DML: {} {}", configFile, JSON.toJSONString(dml, Feature.WriteNulls));
}
DbMapping dbMapping = config.getDbMapping();
if (!dbMapping.isAlter()) {

6
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java
View File

@@ -13,8 +13,8 @@ import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter.Feature;
import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.rdb.config.MirrorDbConfig;
import com.alibaba.otter.canal.client.adapter.rdb.support.SingleDml;
@@ -68,7 +68,7 @@ public class RdbMirrorDbSyncService {
if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) {
// DDL
if (logger.isDebugEnabled()) {
logger.debug("DDL: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
logger.debug("DDL: {}", JSON.toJSONString(dml, Feature.WriteNulls));
}
executeDdl(mirrorDbConfig, dml);
rdbSyncService.getColumnsTypeCache().remove(destination + "." + database + "." + dml.getTable());

6
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java
View File

@@ -21,8 +21,8 @@ import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter.Feature;
import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig.DbMapping;
import com.alibaba.otter.canal.client.adapter.rdb.support.BatchExecutor;
@@ -220,7 +220,7 @@ public class RdbSyncService {
truncate(batchExecutor, config);
}
if (logger.isDebugEnabled()) {
logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
logger.debug("DML: {}", JSON.toJSONString(dml, Feature.WriteNulls));
}
} catch (SQLException e) {
throw new RuntimeException(e);

6
client-adapter/tablestore/src/main/java/com/alibaba/otter/canal/client/adapter/tablestore/service/TablestoreSyncService.java
View File

@@ -6,8 +6,8 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter.Feature;
import com.alibaba.otter.canal.client.adapter.tablestore.enums.TablestoreFieldType;
import com.alibaba.otter.canal.client.adapter.tablestore.support.SyncUtil;
import com.alicloud.openservices.tablestore.TableStoreWriter;
@@ -59,7 +59,7 @@ public class TablestoreSyncService {
Future<WriterResult> future = writer.addRowChangeWithFuture(rowChanges);
if (logger.isDebugEnabled()) {
logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
logger.debug("DML: {}", JSON.toJSONString(dml, Feature.WriteNulls));
}
return future;
}

4
client/pom.xml
View File

@@ -50,10 +50,6 @@
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>

2
client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java
View File

@@ -10,7 +10,7 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.client.CanalMQConnector;
import com.alibaba.otter.canal.client.impl.SimpleCanalConnector;
import com.alibaba.otter.canal.protocol.FlatMessage;

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

@@ -1,6 +1,6 @@
package com.alibaba.otter.canal.client.kafka;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.client.kafka.protocol.KafkaFlatMessage;
import com.alibaba.otter.canal.client.kafka.protocol.KafkaMessage;
import com.alibaba.otter.canal.protocol.FlatMessage;

2
client/src/main/java/com/alibaba/otter/canal/client/pulsarmq/PulsarMQCanalConnector.java
View File

@@ -1,6 +1,6 @@
package com.alibaba.otter.canal.client.pulsarmq;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.client.CanalMQConnector;
import com.alibaba.otter.canal.client.CanalMessageDeserializer;
import com.alibaba.otter.canal.client.impl.SimpleCanalConnector;

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

@@ -1,6 +1,6 @@
package com.alibaba.otter.canal.client.rabbitmq;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.client.CanalMQConnector;
import com.alibaba.otter.canal.client.CanalMessageDeserializer;
import com.alibaba.otter.canal.client.ConsumerBatchMessage;

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

@@ -22,7 +22,7 @@ import org.apache.rocketmq.remoting.RPCHook;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.client.CanalMQConnector;
import com.alibaba.otter.canal.client.CanalMessageDeserializer;
import com.alibaba.otter.canal.client.impl.SimpleCanalConnector;

4
common/pom.xml
View File

@@ -39,8 +39,8 @@
<artifactId>commons-codec</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<groupId>com.alibaba.fastjson2</groupId>
<artifactId>fastjson2</artifactId>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>

55
common/src/main/java/com/alibaba/otter/canal/common/utils/JsonUtils.java
View File

@@ -1,6 +1,5 @@
package com.alibaba.otter.canal.common.utils;
import java.io.IOException;
import java.lang.reflect.Type;
import java.net.Inet4Address;
import java.net.Inet6Address;
@@ -8,14 +7,13 @@ import java.net.InetAddress;
import java.util.Arrays;
import java.util.List;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.alibaba.fastjson.parser.ParserConfig;
import com.alibaba.fastjson.serializer.JSONSerializer;
import com.alibaba.fastjson.serializer.ObjectSerializer;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.alibaba.fastjson.serializer.SerializeWriter;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONFactory;
import com.alibaba.fastjson2.TypeReference;
import com.alibaba.fastjson2.filter.PropertyFilter;
import com.alibaba.fastjson2.JSONWriter;
import com.alibaba.fastjson2.writer.ObjectWriter;
/**
* 字节处理相关工具类
@@ -25,13 +23,12 @@ import com.alibaba.fastjson.serializer.SerializerFeature;
public class JsonUtils {
static {
SerializeConfig.getGlobalInstance().put(InetAddress.class, InetAddressSerializer.instance);
SerializeConfig.getGlobalInstance().put(Inet4Address.class, InetAddressSerializer.instance);
SerializeConfig.getGlobalInstance().put(Inet6Address.class, InetAddressSerializer.instance);
// ParserConfig.getGlobalInstance().setAutoTypeSupport(true);
JSON.register(InetAddress.class, InetAddressWriter.instance);
JSON.register(Inet4Address.class, InetAddressWriter.instance);
JSON.register(Inet6Address.class, InetAddressWriter.instance);
ParserConfig.getGlobalInstance().addAccept("com.alibaba.otter.");
ParserConfig.getGlobalInstance().addAccept("com.taobao.tddl.dbsync.");
JSONFactory.getDefaultObjectReaderProvider().addAutoTypeAccept("com.alibaba.otter.");
JSONFactory.getDefaultObjectReaderProvider().addAutoTypeAccept("com.taobao.tddl.dbsync.");
}
public static <T> T unmarshalFromByte(byte[] bytes, Class<T> targetClass) {
@@ -46,7 +43,7 @@ public class JsonUtils {
return JSON.toJSONBytes(obj); // 默认为UTF-8
}
public static byte[] marshalToByte(Object obj, SerializerFeature... features) {
public static byte[] marshalToByte(Object obj, JSONWriter.Feature... features) {
return JSON.toJSONBytes(obj, features); // 默认为UTF-8
}
@@ -62,7 +59,7 @@ public class JsonUtils {
return JSON.toJSONString(obj); // 默认为UTF-8
}
public static String marshalToString(Object obj, SerializerFeature... features) {
public static String marshalToString(Object obj, JSONWriter.Feature... features) {
return JSON.toJSONString(obj, features); // 默认为UTF-8
}
@@ -71,29 +68,29 @@ public class JsonUtils {
*/
public static String marshalToString(Object obj, String... fliterFields) {
final List<String> propertyFliters = Arrays.asList(fliterFields);
try (SerializeWriter out = new SerializeWriter()) {
JSONSerializer serializer = new JSONSerializer(out);
serializer.getPropertyFilters().add((source, name, value) -> !propertyFliters.contains(name));
serializer.write(obj);
return out.toString();
}
return JSON.toJSONString(obj, new PropertyFilter() {
@Override
public boolean process(Object object, String name, Object value) {
return !propertyFliters.contains(name);
}
});
}
public static class InetAddressSerializer implements ObjectSerializer {
public static class InetAddressWriter implements ObjectWriter {
public static InetAddressSerializer instance = new InetAddressSerializer();
public static InetAddressWriter instance = new InetAddressWriter();
@Override
public void write(JSONSerializer serializer, Object object, Object fieldName, Type fieldType, int features)
throws IOException {
public void write(JSONWriter jsonWriter, Object object, Object fieldName, Type fieldType, long features) {
if (object == null) {
serializer.writeNull();
jsonWriter.writeNull();
return;
}
InetAddress address = (InetAddress) object;
// 优先使用name
serializer.write(address.getHostName());
jsonWriter.writeString(address.getHostName());
}
}
}

2
connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/consumer/CanalKafkaConsumer.java
View File

@@ -16,7 +16,7 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.connector.core.config.CanalConstants;
import com.alibaba.otter.canal.connector.core.consumer.CommonMessage;
import com.alibaba.otter.canal.connector.core.spi.CanalMsgConsumer;

8
connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/producer/CanalKafkaProducer.java
View File

@@ -18,8 +18,8 @@ import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter;
import com.alibaba.otter.canal.common.utils.ExecutorTemplate;
import com.alibaba.otter.canal.connector.core.producer.AbstractMQProducer;
import com.alibaba.otter.canal.connector.core.producer.MQDestination;
@@ -249,13 +249,13 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
FlatMessage flatMessagePart = partitionFlatMessage[i];
if (flatMessagePart != null) {
records.add(new ProducerRecord<>(topicName, i, null, JSON.toJSONBytes(flatMessagePart,
SerializerFeature.WriteMapNullValue)));
JSONWriter.Feature.WriteNulls)));
}
}
} else {
final int partition = mqDestination.getPartition() != null ? mqDestination.getPartition() : 0;
records.add(new ProducerRecord<>(topicName, partition, null, JSON.toJSONBytes(flatMessage,
SerializerFeature.WriteMapNullValue)));
JSONWriter.Feature.WriteNulls)));
}
}
}

2
connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java
View File

@@ -1,6 +1,6 @@
package com.alibaba.otter.canal.connector.pulsarmq.consumer;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.common.utils.MQUtil;
import com.alibaba.otter.canal.connector.core.config.CanalConstants;
import com.alibaba.otter.canal.connector.core.consumer.CommonMessage;

6
connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java
View File

@@ -1,7 +1,7 @@
package com.alibaba.otter.canal.connector.pulsarmq.producer;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter.Feature;
import com.alibaba.otter.canal.common.utils.ExecutorTemplate;
import com.alibaba.otter.canal.common.utils.NamedThreadFactory;
import com.alibaba.otter.canal.common.utils.PropertiesUtils;
@@ -305,7 +305,7 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
MessageId msgResultId = producer
.newMessage()
.property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partition))
.value(JSON.toJSONBytes(f, SerializerFeature.WriteMapNullValue))
.value(JSON.toJSONBytes(f, Feature.WriteNulls))
.send()
//
;

2
connector/pulsarmq-connector/src/test/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumerTest.java
View File

@@ -1,6 +1,6 @@
package com.alibaba.otter.canal.connector.pulsarmq.consumer;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.connector.core.config.CanalConstants;
import com.alibaba.otter.canal.connector.core.consumer.CommonMessage;
import com.alibaba.otter.canal.connector.pulsarmq.config.PulsarMQConstants;

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

@@ -13,7 +13,7 @@ import com.alibaba.otter.canal.common.utils.PropertiesUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.connector.core.config.CanalConstants;
import com.alibaba.otter.canal.connector.core.consumer.CommonMessage;
import com.alibaba.otter.canal.connector.core.spi.CanalMsgConsumer;

6
connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/producer/CanalRabbitMQProducer.java
View File

@@ -10,8 +10,8 @@ import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter;
import com.alibaba.otter.canal.common.CanalException;
import com.alibaba.otter.canal.common.utils.ExecutorTemplate;
import com.alibaba.otter.canal.common.utils.PropertiesUtils;
@@ -151,7 +151,7 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ
// 串行分区
List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(datas, messageSub.getId());
for (FlatMessage flatMessage : flatMessages) {
byte[] message = JSON.toJSONBytes(flatMessage, SerializerFeature.WriteMapNullValue);
byte[] message = JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls);
if (logger.isDebugEnabled()) {
logger.debug("send message:{} to destination:{}", message, canalDestination.getCanalDestination());
}

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

@@ -21,7 +21,7 @@ import org.apache.rocketmq.remoting.RPCHook;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.connector.core.config.CanalConstants;
import com.alibaba.otter.canal.connector.core.consumer.CommonMessage;
import com.alibaba.otter.canal.connector.core.spi.CanalMsgConsumer;

8
connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java
View File

@@ -1,7 +1,7 @@
package com.alibaba.otter.canal.connector.rocketmq.producer;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter;
import com.alibaba.otter.canal.common.CanalException;
import com.alibaba.otter.canal.common.utils.ExecutorTemplate;
import com.alibaba.otter.canal.common.utils.NamedThreadFactory;
@@ -263,7 +263,7 @@ import java.util.stream.Collectors;
List<Message> messages = flatMessagePart.stream()
.map(flatMessage -> new Message(topicName,
((RocketMQProducerConfig) this.mqProperties).getTag(),
JSON.toJSONBytes(flatMessage, SerializerFeature.WriteMapNullValue)))
JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls)))
.collect(Collectors.toList());
// 批量发送
sendMessage(messages, index);
@@ -278,7 +278,7 @@ import java.util.stream.Collectors;
List<Message> messages = flatMessages.stream()
.map(flatMessage -> new Message(topicName,
((RocketMQProducerConfig) this.mqProperties).getTag(),
JSON.toJSONBytes(flatMessage, SerializerFeature.WriteMapNullValue)))
JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls)))
.collect(Collectors.toList());
// 批量发送
sendMessage(messages, partition);

2
instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/HttpHelper.java
View File

@@ -29,7 +29,7 @@ import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSON;
/**
* http client 工具类

6
instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/PlainCanalConfigClient.java
View File

@@ -10,8 +10,8 @@ import java.util.Properties;
import org.apache.commons.lang.StringUtils;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.common.CanalException;
import com.alibaba.otter.canal.common.CanalLifeCycle;
@@ -119,7 +119,7 @@ public class PlainCanalConfigClient extends AbstractCanalLifeCycle implements Ca
heads.put("user", user);
heads.put("passwd", passwd);
String response = httpHelper.get(url, heads, REQUEST_TIMEOUT);
ResponseModel<CanalConfig> resp = JSONObject.parseObject(response,
ResponseModel<CanalConfig> resp = JSON.parseObject(response,
new TypeReference<ResponseModel<CanalConfig>>() {
});

8
meta/src/main/java/com/alibaba/otter/canal/meta/ZooKeeperMetaManager.java
View File

@@ -12,7 +12,7 @@ import org.apache.commons.lang.StringUtils;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSONWriter;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.common.utils.JsonUtils;
import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
@@ -155,7 +155,7 @@ public class ZooKeeperMetaManager extends AbstractCanalLifeCycle implements Cana
public void updateCursor(ClientIdentity clientIdentity, Position position) throws CanalMetaManagerException {
String path = ZookeeperPathUtils.getCursorPath(clientIdentity.getDestination(), clientIdentity.getClientId());
byte[] data = JsonUtils.marshalToByte(position, SerializerFeature.WriteClassName);
byte[] data = JsonUtils.marshalToByte(position, JSONWriter.Feature.WriteClassName);
try {
zkClientx.writeData(path, data);
} catch (ZkNoNodeException e) {
@@ -166,7 +166,7 @@ public class ZooKeeperMetaManager extends AbstractCanalLifeCycle implements Cana
public Long addBatch(ClientIdentity clientIdentity, PositionRange positionRange) throws CanalMetaManagerException {
String path = ZookeeperPathUtils.getBatchMarkPath(clientIdentity.getDestination(),
clientIdentity.getClientId());
byte[] data = JsonUtils.marshalToByte(positionRange, SerializerFeature.WriteClassName);
byte[] data = JsonUtils.marshalToByte(positionRange, JSONWriter.Feature.WriteClassName);
String batchPath = zkClientx
.createPersistentSequential(path + ZookeeperPathUtils.ZOOKEEPER_SEPARATOR, data, true);
String batchIdString = StringUtils.substringAfterLast(batchPath, ZookeeperPathUtils.ZOOKEEPER_SEPARATOR);
@@ -177,7 +177,7 @@ public class ZooKeeperMetaManager extends AbstractCanalLifeCycle implements Cana
Long batchId) throws CanalMetaManagerException {
String path = ZookeeperPathUtils
.getBatchMarkWithIdPath(clientIdentity.getDestination(), clientIdentity.getClientId(), batchId);
byte[] data = JsonUtils.marshalToByte(positionRange, SerializerFeature.WriteClassName);
byte[] data = JsonUtils.marshalToByte(positionRange, JSONWriter.Feature.WriteClassName);
zkClientx.createPersistent(path, data, true);
}

4
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBackupPolicyRequest.java
View File

@@ -3,8 +3,8 @@ package com.alibaba.otter.canal.parse.inbound.mysql.rds.request;
import org.apache.http.HttpResponse;
import org.apache.http.util.EntityUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.RdsBackupPolicy;
/**

6
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBinlogFilesRequest.java
View File

@@ -2,11 +2,11 @@ package com.alibaba.otter.canal.parse.inbound.mysql.rds.request;
import java.util.Date;
import com.alibaba.fastjson2.JSON;
import org.apache.http.HttpResponse;
import org.apache.http.util.EntityUtils;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.alibaba.fastjson2.TypeReference;
import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.DescribeBinlogFileResult;
/**
@@ -47,7 +47,7 @@ public class DescribeBinlogFilesRequest extends AbstractRequest<DescribeBinlogFi
@Override
protected DescribeBinlogFileResult processResult(HttpResponse response) throws Exception {
String result = EntityUtils.toString(response.getEntity());
DescribeBinlogFileResult describeBinlogFileResult = JSONObject.parseObject(result,
DescribeBinlogFileResult describeBinlogFileResult = JSON.parseObject(result,
new TypeReference<DescribeBinlogFileResult>() {
});
return describeBinlogFileResult;

4
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java
View File

@@ -22,8 +22,8 @@ import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import com.alibaba.druid.sql.repository.Schema;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.otter.canal.filter.CanalEventFilter;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket;
import com.alibaba.otter.canal.parse.exception.CanalParseException;

6
pom.xml
View File

@@ -218,9 +218,9 @@
<version>0.10</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.58.sec06</version>
<groupId>com.alibaba.fastjson2</groupId>
<artifactId>fastjson2</artifactId>
<version>2.0.2</version>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>