Merge branch '1.10_release_4.2.x' into 1.10_release · DTStack/flinkStreamSQL@1609917 · GitHub
Skip to content

Commit 1609917

Browse files
FlechazoWFlechazoW
committed
Merge branch '1.10_release_4.2.x' into 1.10_release
# Conflicts: # core/src/main/java/com/dtstack/flink/sql/table/AbstractTableInfo.java # core/src/main/java/com/dtstack/flink/sql/util/DtStringUtil.java # elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/Es6Util.java # impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaOutputFormat.java # pom.xml # rdb/rdb-core/src/main/java/com/dtstack/flink/sql/core/rdb/JdbcResourceCheck.java # rdb/rdb-core/src/main/java/com/dtstack/flink/sql/core/rdb/util/JdbcConnectionUtil.java # rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/JDBCUpsertOutputFormat.java # rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/table/RdbSinkParser.java
2 parents 0cda3c4 + cc32339 commit 1609917

68 files changed

Lines changed: 2534 additions & 261 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java‎

Lines changed: 2 additions & 3 deletions

‎core/src/main/java/com/dtstack/flink/sql/resource/ResourceCheck.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@
2929
* @description:资源检测
3030
**/
3131
public abstract class ResourceCheck {
32-
public static Boolean NEED_CHECK = true;
32+
public static Boolean NEED_CHECK = false;
3333
public static String CHECK_STR = "checkResource";
3434
protected static Logger LOG = LoggerFactory.getLogger(ResourceCheck.class);
3535
public String TABLE_TYPE_KEY = "tableType";

‎core/src/main/java/com/dtstack/flink/sql/table/AbstractTableInfo.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,15 +21,13 @@
2121
import com.dtstack.flink.sql.dirtyManager.manager.DirtyKeys;
2222
import com.dtstack.flink.sql.outputformat.AbstractDtRichOutputFormat;
2323
import com.dtstack.flink.sql.side.AbstractSideTableInfo;
24-
2524
import com.google.common.collect.Lists;
2625
import com.google.common.collect.Maps;
2726

2827
import java.io.Serializable;
2928
import java.util.List;
3029
import java.util.Map;
3130
import java.util.Objects;
32-
import java.util.Properties;
3331

3432
/**
3533
* Reason:
@@ -43,6 +41,9 @@ public abstract class AbstractTableInfo implements Serializable {
4341

4442
public static final String PARALLELISM_KEY = "parallelism";
4543
public static final String ERROR_LIMIT = "errorLimit";
44+
public static final Boolean DEFAULT_FALSE = false;
45+
public static final Boolean DEFAULT_TRUE = true;
46+
public static final Object DEFAULT_NULL = null;
4647
private final List<String> fieldList = Lists.newArrayList();
4748
private final List<String> fieldTypeList = Lists.newArrayList();
4849
private final List<Class> fieldClassList = Lists.newArrayList();

‎core/src/main/java/com/dtstack/flink/sql/util/DtStringUtil.java‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -21,10 +21,11 @@
2121
package com.dtstack.flink.sql.util;
2222

2323
import com.dtstack.flink.sql.enums.ColumnType;
24-
import com.fasterxml.jackson.databind.ObjectMapper;
2524
import com.google.common.base.Strings;
2625
import com.google.common.collect.Maps;
2726
import org.apache.commons.lang3.StringUtils;
27+
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
28+
import org.apache.flink.util.Preconditions;
2829

2930
import java.math.BigDecimal;
3031
import java.sql.Timestamp;
@@ -46,7 +47,7 @@ public class DtStringUtil {
4647

4748
private static final Pattern NO_VERSION_PATTERN = Pattern.compile("([a-zA-Z]+).*");
4849

49-
private static ObjectMapper objectMapper = new ObjectMapper();
50+
private static final ObjectMapper objectMapper = new ObjectMapper();
5051

5152
/**
5253
* Split the specified string delimiter --- ignored quotes delimiter
@@ -255,10 +256,12 @@ public static String col2string(Object column, String type) {
255256
return result.toString();
256257
}
257258

258-
public static String getPluginTypeWithoutVersion(String engineType){
259+
public static String getPluginTypeWithoutVersion(String engineType) {
260+
Preconditions.checkNotNull(engineType, "type can't be null!");
259261

260262
Matcher matcher = NO_VERSION_PATTERN.matcher(engineType);
261-
if(!matcher.find()){
263+
264+
if (!matcher.find()) {
262265
return engineType;
263266
}
264267

‎core/src/main/java/com/dtstack/flink/sql/util/MathUtil.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -263,4 +263,16 @@ public static Timestamp getTimestamp(Object obj) {
263263
throw new RuntimeException("not support type of " + obj.getClass() + " convert to Date.");
264264
}
265265

266+
public static Character getChar(Object obj) {
267+
if (obj == null) {
268+
return null;
269+
}
270+
if (obj instanceof Character) {
271+
return (Character) obj;
272+
}
273+
if (obj instanceof String) {
274+
return String.valueOf(obj).charAt(0);
275+
}
276+
throw new RuntimeException("not support type of " + obj.getClass() + " convert to Char");
277+
}
266278
}

‎core/src/main/java/com/dtstack/flink/sql/util/PluginUtil.java‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -22,11 +22,9 @@
2222

2323
import com.dtstack.flink.sql.dirtyManager.consumer.DirtyConsumerFactory;
2424
import com.dtstack.flink.sql.enums.EPluginLoadMode;
25-
import com.fasterxml.jackson.core.JsonGenerationException;
26-
import com.fasterxml.jackson.core.JsonParseException;
27-
import com.fasterxml.jackson.databind.JsonMappingException;
28-
import com.fasterxml.jackson.databind.ObjectMapper;
2925
import org.apache.commons.lang3.StringUtils;
26+
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
27+
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
3028
import org.slf4j.Logger;
3129
import org.slf4j.LoggerFactory;
3230

@@ -128,7 +126,11 @@ public static Map<String,Object> objectToMap(Object obj) throws Exception{
128126
return objectMapper.readValue(objectMapper.writeValueAsBytes(obj), Map.class);
129127
}
130128

131-
public static <T> T jsonStrToObject(String jsonStr, Class<T> clazz) throws JsonParseException, JsonMappingException, JsonGenerationException, IOException{
129+
public static String objectToString(Object obj) throws JsonProcessingException {
130+
return objectMapper.writeValueAsString(obj);
131+
}
132+
133+
public static <T> T jsonStrToObject(String jsonStr, Class<T> clazz) throws IOException{
132134
return objectMapper.readValue(jsonStr, clazz);
133135
}
134136

‎core/src/main/java/com/dtstack/flink/sql/util/ThreadUtil.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
*/
2828
public class ThreadUtil {
2929
public static final Long DEFAULT_SLEEP_TIME = 10L;
30+
3031
public static void sleepSeconds(long timeout) {
3132
try {
3233
TimeUnit.SECONDS.sleep(timeout);
@@ -50,4 +51,12 @@ public static void sleepMicroseconds(long timeout) {
5051
throw new RuntimeException(ie);
5152
}
5253
}
54+
55+
public static void sleepMilliseconds(long timeout) {
56+
try {
57+
TimeUnit.MILLISECONDS.sleep(timeout);
58+
} catch (InterruptedException e) {
59+
throw new RuntimeException(e);
60+
}
61+
}
5362
}

‎docs/plugin/filesource.md‎

Lines changed: 184 additions & 0 deletions

0 commit comments

Comments
 (0)