Skip to content

D
dlink
Commit f6d81b81 authored by wenmo's avatar wenmo
Browse files
Options

client format

parent 2a471036
Hide whitespace changes
Inline Side-by-side
Showing with 483 additions and 398 deletions
dlink-client/dlink-client-1.11/src/main/java/com/dlink/cdc/FlinkCDCMergeBuilder.java
View file @ f6d81b81
...@@ -21,10 +21,10 @@ import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer; ...@@ -21,10 +21,10 @@ import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
public class FlinkCDCMergeBuilder { public class FlinkCDCMergeBuilder {
public static void buildMySqlCDC(StreamExecutionEnvironment env, FlinkCDCConfig config) { public static void buildMySqlCDC(StreamExecutionEnvironment env, FlinkCDCConfig config) {
if(Asserts.isNotNull(config.getParallelism())){ if (Asserts.isNotNull(config.getParallelism())) {
env.setParallelism(config.getParallelism()); env.setParallelism(config.getParallelism());
} }
if(Asserts.isNotNull(config.getCheckpoint())){ if (Asserts.isNotNull(config.getCheckpoint())) {
env.enableCheckpointing(config.getCheckpoint()); env.enableCheckpointing(config.getCheckpoint());
} }
MySqlSourceBuilder<String> sourceBuilder = MySqlSource.<String>builder() MySqlSourceBuilder<String> sourceBuilder = MySqlSource.<String>builder()
...@@ -32,16 +32,16 @@ public class FlinkCDCMergeBuilder { ...@@ -32,16 +32,16 @@ public class FlinkCDCMergeBuilder {
.port(config.getPort()) .port(config.getPort())
.username(config.getUsername()) .username(config.getUsername())
.password(config.getPassword()); .password(config.getPassword());
if(Asserts.isNotNull(config.getDatabase())&&config.getDatabase().size()>0){ if (Asserts.isNotNull(config.getDatabase()) && config.getDatabase().size() > 0) {
sourceBuilder.databaseList(config.getDatabase().toArray(new String[0])); sourceBuilder.databaseList(config.getDatabase().toArray(new String[0]));
} }
if(Asserts.isNotNull(config.getTable())&&config.getTable().size()>0){ if (Asserts.isNotNull(config.getTable()) && config.getTable().size() > 0) {
sourceBuilder.tableList(config.getTable().toArray(new String[0])); sourceBuilder.tableList(config.getTable().toArray(new String[0]));
} }
MySqlSourceBuilder<String> builder = sourceBuilder MySqlSourceBuilder<String> builder = sourceBuilder
.deserializer(new JsonDebeziumDeserializationSchema()); .deserializer(new JsonDebeziumDeserializationSchema());
if(Asserts.isNotNullString(config.getStartupMode())){ if (Asserts.isNotNullString(config.getStartupMode())) {
switch (config.getStartupMode().toUpperCase()){ switch (config.getStartupMode().toUpperCase()) {
case "INITIAL": case "INITIAL":
builder.startupOptions(StartupOptions.initial()); builder.startupOptions(StartupOptions.initial());
break; break;
...@@ -54,12 +54,12 @@ public class FlinkCDCMergeBuilder { ...@@ -54,12 +54,12 @@ public class FlinkCDCMergeBuilder {
default: default:
builder.startupOptions(StartupOptions.latest()); builder.startupOptions(StartupOptions.latest());
} }
}else { } else {
builder.startupOptions(StartupOptions.latest()); builder.startupOptions(StartupOptions.latest());
} }
MySqlSource<String> sourceFunction = builder.build(); MySqlSource<String> sourceFunction = builder.build();
DataStreamSource<String> streamSource = env.fromSource(sourceFunction, WatermarkStrategy.noWatermarks(), "MySQL Source"); DataStreamSource<String> streamSource = env.fromSource(sourceFunction, WatermarkStrategy.noWatermarks(), "MySQL Source");
streamSource.addSink(getKafkaProducer(config.getBrokers(),config.getTopic())); streamSource.addSink(getKafkaProducer(config.getBrokers(), config.getTopic()));
} }
private static FlinkKafkaProducer<String> getKafkaProducer(String brokers, String topic) { private static FlinkKafkaProducer<String> getKafkaProducer(String brokers, String topic) {
......
dlink-client/dlink-client-1.11/src/main/java/com/dlink/executor/CustomTableResultImpl.java
View file @ f6d81b81
...@@ -17,8 +17,9 @@ import java.util.Optional; ...@@ -17,8 +17,9 @@ import java.util.Optional;
/** /**
* 定制TableResultImpl * 定制TableResultImpl
* @author wenmo *
* @since 2021/6/7 22:06 * @author wenmo
* @since 2021/6/7 22:06
**/ **/
@Internal @Internal
class CustomTableResultImpl implements TableResult { class CustomTableResultImpl implements TableResult {
...@@ -49,12 +50,12 @@ class CustomTableResultImpl implements TableResult { ...@@ -49,12 +50,12 @@ class CustomTableResultImpl implements TableResult {
this.printStyle = Preconditions.checkNotNull(printStyle, "printStyle should not be null"); this.printStyle = Preconditions.checkNotNull(printStyle, "printStyle should not be null");
} }
public static TableResult buildTableResult(List<TableSchemaField> fields,List<Row> rows){ public static TableResult buildTableResult(List<TableSchemaField> fields, List<Row> rows) {
Builder builder = builder().resultKind(ResultKind.SUCCESS); Builder builder = builder().resultKind(ResultKind.SUCCESS);
if(fields.size()>0) { if (fields.size() > 0) {
TableSchema.Builder tableSchemaBuild = TableSchema.builder(); TableSchema.Builder tableSchemaBuild = TableSchema.builder();
for (int i = 0; i < fields.size(); i++) { for (int i = 0; i < fields.size(); i++) {
tableSchemaBuild.field(fields.get(i).getName(),fields.get(i).getType()); tableSchemaBuild.field(fields.get(i).getName(), fields.get(i).getType());
} }
builder.tableSchema(tableSchemaBuild.build()).data(rows); builder.tableSchema(tableSchemaBuild.build()).data(rows);
} }
...@@ -109,7 +110,9 @@ class CustomTableResultImpl implements TableResult { ...@@ -109,7 +110,9 @@ class CustomTableResultImpl implements TableResult {
return new Builder(); return new Builder();
} }
/** Builder for creating a {@link CustomTableResultImpl}. */ /**
* Builder for creating a {@link CustomTableResultImpl}.
*/
public static class Builder { public static class Builder {
private JobClient jobClient = null; private JobClient jobClient = null;
private TableSchema tableSchema = null; private TableSchema tableSchema = null;
...@@ -118,7 +121,8 @@ class CustomTableResultImpl implements TableResult { ...@@ -118,7 +121,8 @@ class CustomTableResultImpl implements TableResult {
private PrintStyle printStyle = private PrintStyle printStyle =
PrintStyle.tableau(Integer.MAX_VALUE, PrintUtils.NULL_COLUMN, false); PrintStyle.tableau(Integer.MAX_VALUE, PrintUtils.NULL_COLUMN, false);
private Builder() {} private Builder() {
}
/** /**
* Specifies job client which associates the submitted Flink job. * Specifies job client which associates the submitted Flink job.
...@@ -174,20 +178,26 @@ class CustomTableResultImpl implements TableResult { ...@@ -174,20 +178,26 @@ class CustomTableResultImpl implements TableResult {
return this; return this;
} }
/** Specifies print style. Default is {@link TableauStyle} with max integer column width. */ /**
* Specifies print style. Default is {@link TableauStyle} with max integer column width.
*/
public Builder setPrintStyle(PrintStyle printStyle) { public Builder setPrintStyle(PrintStyle printStyle) {
Preconditions.checkNotNull(printStyle, "printStyle should not be null"); Preconditions.checkNotNull(printStyle, "printStyle should not be null");
this.printStyle = printStyle; this.printStyle = printStyle;
return this; return this;
} }
/** Returns a {@link TableResult} instance. */ /**
* Returns a {@link TableResult} instance.
*/
public TableResult build() { public TableResult build() {
return new CustomTableResultImpl(jobClient, tableSchema, resultKind, data, printStyle); return new CustomTableResultImpl(jobClient, tableSchema, resultKind, data, printStyle);
} }
} }
/** Root interface for all print styles. */ /**
* Root interface for all print styles.
*/
public interface PrintStyle { public interface PrintStyle {
/** /**
* Create a tableau print style with given max column width, null column, and a flag to * Create a tableau print style with given max column width, null column, and a flag to
...@@ -211,7 +221,9 @@ class CustomTableResultImpl implements TableResult { ...@@ -211,7 +221,9 @@ class CustomTableResultImpl implements TableResult {
} }
} }
/** print the result schema and content as tableau form. */ /**
* print the result schema and content as tableau form.
*/
private static final class TableauStyle implements PrintStyle { private static final class TableauStyle implements PrintStyle {
/** /**
* A flag to indicate whether the column width is derived from type (true) or content * A flag to indicate whether the column width is derived from type (true) or content
...@@ -245,5 +257,6 @@ class CustomTableResultImpl implements TableResult { ...@@ -245,5 +257,6 @@ class CustomTableResultImpl implements TableResult {
/** /**
* only print the result content as raw form. column delimiter is ",", row delimiter is "\n". * only print the result content as raw form. column delimiter is ",", row delimiter is "\n".
*/ */
private static final class RawContentStyle implements PrintStyle {} private static final class RawContentStyle implements PrintStyle {
}
} }
dlink-client/dlink-client-1.11/src/main/java/com/dlink/executor/TableSchemaField.java
View file @ f6d81b81
...@@ -3,8 +3,8 @@ package com.dlink.executor; ...@@ -3,8 +3,8 @@ package com.dlink.executor;
import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.DataType;
/** /**
* @author wenmo * @author wenmo
* @since 2021/6/7 22:06 * @since 2021/6/7 22:06
**/ **/
public class TableSchemaField { public class TableSchemaField {
private String name; private String name;
......
dlink-client/dlink-client-1.11/src/main/java/com/dlink/utils/FlinkUtil.java
View file @ f6d81b81
...@@ -17,18 +17,18 @@ import java.util.Optional; ...@@ -17,18 +17,18 @@ import java.util.Optional;
*/ */
public class FlinkUtil { public class FlinkUtil {
public static List<String> getFieldNamesFromCatalogManager(CatalogManager catalogManager, String catalog, String database, String table){ public static List<String> getFieldNamesFromCatalogManager(CatalogManager catalogManager, String catalog, String database, String table) {
Optional<CatalogManager.TableLookupResult> tableOpt = catalogManager.getTable( Optional<CatalogManager.TableLookupResult> tableOpt = catalogManager.getTable(
ObjectIdentifier.of(catalog, database, table) ObjectIdentifier.of(catalog, database, table)
); );
if (tableOpt.isPresent()) { if (tableOpt.isPresent()) {
return Arrays.asList(tableOpt.get().getResolvedSchema().getFieldNames()); return Arrays.asList(tableOpt.get().getResolvedSchema().getFieldNames());
}else{ } else {
return new ArrayList<String>(); return new ArrayList<String>();
} }
} }
public static List<String> catchColumn(TableResult tableResult){ public static List<String> catchColumn(TableResult tableResult) {
return Arrays.asList(tableResult.getTableSchema().getFieldNames()); return Arrays.asList(tableResult.getTableSchema().getFieldNames());
} }
} }
dlink-client/dlink-client-1.11/src/main/java/org/apache/flink/table/types/extraction/ExtractionUtils.java
View file @ f6d81b81
...@@ -21,48 +21,28 @@ package org.apache.flink.table.types.extraction; ...@@ -21,48 +21,28 @@ package org.apache.flink.table.types.extraction;
import com.dlink.pool.ClassPool; import com.dlink.pool.ClassPool;
import org.apache.flink.annotation.Internal; import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.shaded.asm7.org.objectweb.asm.*;
import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.DataTypeFactory; import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.StructuredType; import org.apache.flink.table.types.logical.StructuredType;
import org.apache.flink.shaded.asm7.org.objectweb.asm.ClassReader;
import org.apache.flink.shaded.asm7.org.objectweb.asm.ClassVisitor;
import org.apache.flink.shaded.asm7.org.objectweb.asm.Label;
import org.apache.flink.shaded.asm7.org.objectweb.asm.MethodVisitor;
import org.apache.flink.shaded.asm7.org.objectweb.asm.Opcodes;
import javax.annotation.Nullable; import javax.annotation.Nullable;
import java.io.IOException; import java.io.IOException;
import java.lang.annotation.Annotation; import java.lang.annotation.Annotation;
import java.lang.reflect.Constructor;
import java.lang.reflect.Executable;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.lang.reflect.Parameter;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type; import java.lang.reflect.Type;
import java.lang.reflect.TypeVariable; import java.lang.reflect.*;
import java.util.ArrayList; import java.util.*;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import java.util.stream.Stream; import java.util.stream.Stream;
import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getConstructorDescriptor; import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getConstructorDescriptor;
import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getMethodDescriptor; import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getMethodDescriptor;
/** Utilities for performing reflection tasks. */ /**
* Utilities for performing reflection tasks.
*/
@Internal @Internal
public final class ExtractionUtils { public final class ExtractionUtils {
...@@ -70,7 +50,9 @@ public final class ExtractionUtils { ...@@ -70,7 +50,9 @@ public final class ExtractionUtils {
// Methods shared across packages // Methods shared across packages
// -------------------------------------------------------------------------------------------- // --------------------------------------------------------------------------------------------
/** Collects methods of the given name. */ /**
* Collects methods of the given name.
*/
public static List<Method> collectMethods(Class<?> function, String methodName) { public static List<Method> collectMethods(Class<?> function, String methodName) {
return Arrays.stream(function.getMethods()) return Arrays.stream(function.getMethods())
.filter(method -> method.getName().equals(methodName)) .filter(method -> method.getName().equals(methodName))
...@@ -130,7 +112,9 @@ public final class ExtractionUtils { ...@@ -130,7 +112,9 @@ public final class ExtractionUtils {
return clz == null || ExtractionUtils.isAssignable(clz, param, true); return clz == null || ExtractionUtils.isAssignable(clz, param, true);
} }
/** Creates a method signature string like {@code int eval(Integer, String)}. */ /**
* Creates a method signature string like {@code int eval(Integer, String)}.
*/
public static String createMethodSignatureString( public static String createMethodSignatureString(
String methodName, Class<?>[] parameters, @Nullable Class<?> returnType) { String methodName, Class<?>[] parameters, @Nullable Class<?> returnType) {
final StringBuilder builder = new StringBuilder(); final StringBuilder builder = new StringBuilder();
...@@ -299,7 +283,9 @@ public final class ExtractionUtils { ...@@ -299,7 +283,9 @@ public final class ExtractionUtils {
return false; return false;
} }
/** Checks whether a field is directly readable without a getter. */ /**
* Checks whether a field is directly readable without a getter.
*/
public static boolean isStructuredFieldDirectlyReadable(Field field) { public static boolean isStructuredFieldDirectlyReadable(Field field) {
final int m = field.getModifiers(); final int m = field.getModifiers();
...@@ -307,7 +293,9 @@ public final class ExtractionUtils { ...@@ -307,7 +293,9 @@ public final class ExtractionUtils {
return Modifier.isPublic(m); return Modifier.isPublic(m);
} }
/** Checks whether a field is directly writable without a setter or constructor. */ /**
* Checks whether a field is directly writable without a setter or constructor.
*/
public static boolean isStructuredFieldDirectlyWritable(Field field) { public static boolean isStructuredFieldDirectlyWritable(Field field) {
final int m = field.getModifiers(); final int m = field.getModifiers();
...@@ -324,12 +312,16 @@ public final class ExtractionUtils { ...@@ -324,12 +312,16 @@ public final class ExtractionUtils {
// Methods intended for this package // Methods intended for this package
// -------------------------------------------------------------------------------------------- // --------------------------------------------------------------------------------------------
/** Helper method for creating consistent exceptions during extraction. */ /**
* Helper method for creating consistent exceptions during extraction.
*/
static ValidationException extractionError(String message, Object... args) { static ValidationException extractionError(String message, Object... args) {
return extractionError(null, message, args); return extractionError(null, message, args);
} }
/** Helper method for creating consistent exceptions during extraction. */ /**
* Helper method for creating consistent exceptions during extraction.
*/
static ValidationException extractionError(Throwable cause, String message, Object... args) { static ValidationException extractionError(Throwable cause, String message, Object... args) {
return new ValidationException(String.format(message, args), cause); return new ValidationException(String.format(message, args), cause);
} }
...@@ -358,8 +350,11 @@ public final class ExtractionUtils { ...@@ -358,8 +350,11 @@ public final class ExtractionUtils {
return typeHierarchy; return typeHierarchy;
} }
/** Converts a {@link Type} to {@link Class} if possible, {@code null} otherwise. */ /**
static @Nullable Class<?> toClass(Type type) { * Converts a {@link Type} to {@link Class} if possible, {@code null} otherwise.
*/
static @Nullable
Class<?> toClass(Type type) {
if (type instanceof Class) { if (type instanceof Class) {
return (Class<?>) type; return (Class<?>) type;
} else if (type instanceof ParameterizedType) { } else if (type instanceof ParameterizedType) {
...@@ -370,7 +365,9 @@ public final class ExtractionUtils { ...@@ -370,7 +365,9 @@ public final class ExtractionUtils {
return null; return null;
} }
/** Creates a raw data type. */ /**
* Creates a raw data type.
*/
@SuppressWarnings({"unchecked", "rawtypes"}) @SuppressWarnings({"unchecked", "rawtypes"})
static DataType createRawType( static DataType createRawType(
DataTypeFactory typeFactory, DataTypeFactory typeFactory,
...@@ -404,7 +401,9 @@ public final class ExtractionUtils { ...@@ -404,7 +401,9 @@ public final class ExtractionUtils {
} }
} }
/** Resolves a {@link TypeVariable} using the given type hierarchy if possible. */ /**
* Resolves a {@link TypeVariable} using the given type hierarchy if possible.
*/
static Type resolveVariable(List<Type> typeHierarchy, TypeVariable<?> variable) { static Type resolveVariable(List<Type> typeHierarchy, TypeVariable<?> variable) {
// iterate through hierarchy from top to bottom until type variable gets a non-variable // iterate through hierarchy from top to bottom until type variable gets a non-variable
// assigned // assigned
...@@ -427,7 +426,8 @@ public final class ExtractionUtils { ...@@ -427,7 +426,8 @@ public final class ExtractionUtils {
return variable; return variable;
} }
private static @Nullable Type resolveVariableInParameterizedType( private static @Nullable
Type resolveVariableInParameterizedType(
TypeVariable<?> variable, ParameterizedType currentType) { TypeVariable<?> variable, ParameterizedType currentType) {
final Class<?> currentRaw = (Class<?>) currentType.getRawType(); final Class<?> currentRaw = (Class<?>) currentType.getRawType();
final TypeVariable<?>[] currentVariables = currentRaw.getTypeParameters(); final TypeVariable<?>[] currentVariables = currentRaw.getTypeParameters();
...@@ -465,7 +465,9 @@ public final class ExtractionUtils { ...@@ -465,7 +465,9 @@ public final class ExtractionUtils {
} }
} }
/** Returns the fields of a class for a {@link StructuredType}. */ /**
* Returns the fields of a class for a {@link StructuredType}.
*/
static List<Field> collectStructuredFields(Class<?> clazz) { static List<Field> collectStructuredFields(Class<?> clazz) {
final List<Field> fields = new ArrayList<>(); final List<Field> fields = new ArrayList<>();
while (clazz != Object.class) { while (clazz != Object.class) {
...@@ -482,7 +484,9 @@ public final class ExtractionUtils { ...@@ -482,7 +484,9 @@ public final class ExtractionUtils {
return fields; return fields;
} }
/** Validates if a field is properly readable either directly or through a getter. */ /**
* Validates if a field is properly readable either directly or through a getter.
*/
static void validateStructuredFieldReadability(Class<?> clazz, Field field) { static void validateStructuredFieldReadability(Class<?> clazz, Field field) {
// field is accessible // field is accessible
if (isStructuredFieldDirectlyReadable(field)) { if (isStructuredFieldDirectlyReadable(field)) {
...@@ -525,7 +529,9 @@ public final class ExtractionUtils { ...@@ -525,7 +529,9 @@ public final class ExtractionUtils {
field.getName(), clazz.getName()); field.getName(), clazz.getName());
} }
/** Returns the boxed type of a primitive type. */ /**
* Returns the boxed type of a primitive type.
*/
static Type primitiveToWrapper(Type type) { static Type primitiveToWrapper(Type type) {
if (type instanceof Class) { if (type instanceof Class) {
return primitiveToWrapper((Class<?>) type); return primitiveToWrapper((Class<?>) type);
...@@ -533,7 +539,9 @@ public final class ExtractionUtils { ...@@ -533,7 +539,9 @@ public final class ExtractionUtils {
return type; return type;
} }
/** Collects all methods that qualify as methods of a {@link StructuredType}. */ /**
* Collects all methods that qualify as methods of a {@link StructuredType}.
*/
static List<Method> collectStructuredMethods(Class<?> clazz) { static List<Method> collectStructuredMethods(Class<?> clazz) {
final List<Method> methods = new ArrayList<>(); final List<Method> methods = new ArrayList<>();
while (clazz != Object.class) { while (clazz != Object.class) {
...@@ -584,7 +592,9 @@ public final class ExtractionUtils { ...@@ -584,7 +592,9 @@ public final class ExtractionUtils {
// Parameter Extraction Utilities // Parameter Extraction Utilities
// -------------------------------------------------------------------------------------------- // --------------------------------------------------------------------------------------------
/** Result of the extraction in {@link #extractAssigningConstructor(Class, List)}. */ /**
* Result of the extraction in {@link #extractAssigningConstructor(Class, List)}.
*/
static class AssigningConstructor { static class AssigningConstructor {
public final Constructor<?> constructor; public final Constructor<?> constructor;
public final List<String> parameterNames; public final List<String> parameterNames;
...@@ -599,7 +609,8 @@ public final class ExtractionUtils { ...@@ -599,7 +609,8 @@ public final class ExtractionUtils {
* Checks whether the given constructor takes all of the given fields with matching (possibly * Checks whether the given constructor takes all of the given fields with matching (possibly
* primitive) type and name. An assigning constructor can define the order of fields. * primitive) type and name. An assigning constructor can define the order of fields.
*/ */
static @Nullable AssigningConstructor extractAssigningConstructor( static @Nullable
AssigningConstructor extractAssigningConstructor(
Class<?> clazz, List<Field> fields) { Class<?> clazz, List<Field> fields) {
AssigningConstructor foundConstructor = null; AssigningConstructor foundConstructor = null;
for (Constructor<?> constructor : clazz.getDeclaredConstructors()) { for (Constructor<?> constructor : clazz.getDeclaredConstructors()) {
...@@ -623,8 +634,11 @@ public final class ExtractionUtils { ...@@ -623,8 +634,11 @@ public final class ExtractionUtils {
return foundConstructor; return foundConstructor;
} }
/** Extracts the parameter names of a method if possible. */ /**
static @Nullable List<String> extractMethodParameterNames(Method method) { * Extracts the parameter names of a method if possible.
*/
static @Nullable
List<String> extractMethodParameterNames(Method method) {
return extractExecutableNames(method); return extractExecutableNames(method);
} }
...@@ -632,7 +646,8 @@ public final class ExtractionUtils { ...@@ -632,7 +646,8 @@ public final class ExtractionUtils {
* Extracts ordered parameter names from a constructor that takes all of the given fields with * Extracts ordered parameter names from a constructor that takes all of the given fields with
* matching (possibly primitive) type and name. * matching (possibly primitive) type and name.
*/ */
private static @Nullable List<String> extractConstructorParameterNames( private static @Nullable
List<String> extractConstructorParameterNames(
Constructor<?> constructor, List<Field> fields) { Constructor<?> constructor, List<Field> fields) {
final Type[] parameterTypes = constructor.getGenericParameterTypes(); final Type[] parameterTypes = constructor.getGenericParameterTypes();
...@@ -659,7 +674,8 @@ public final class ExtractionUtils { ...@@ -659,7 +674,8 @@ public final class ExtractionUtils {
return parameterNames; return parameterNames;
} }
private static @Nullable List<String> extractExecutableNames(Executable executable) { private static @Nullable
List<String> extractExecutableNames(Executable executable) {
final int offset; final int offset;
if (!Modifier.isStatic(executable.getModifiers())) { if (!Modifier.isStatic(executable.getModifiers())) {
// remove "this" as first parameter // remove "this" as first parameter
...@@ -705,7 +721,7 @@ public final class ExtractionUtils { ...@@ -705,7 +721,7 @@ public final class ExtractionUtils {
private static ClassReader getClassReader(Class<?> cls) { private static ClassReader getClassReader(Class<?> cls) {
final String className = cls.getName().replaceFirst("^.*\\.", "") + ".class"; final String className = cls.getName().replaceFirst("^.*\\.", "") + ".class";
if(ClassPool.exist(cls.getName())){ if (ClassPool.exist(cls.getName())) {
return new ClassReader(ClassPool.get(cls.getName()).getClassByte()); return new ClassReader(ClassPool.get(cls.getName()).getClassByte());
} }
try { try {
...@@ -798,8 +814,8 @@ public final class ExtractionUtils { ...@@ -798,8 +814,8 @@ public final class ExtractionUtils {
* href="http://docs.oracle.com/javase/specs/">The Java Language Specification</a></em>, * href="http://docs.oracle.com/javase/specs/">The Java Language Specification</a></em>,
* sections 5.1.1, 5.1.2 and 5.1.4 for details. * sections 5.1.1, 5.1.2 and 5.1.4 for details.
* *
* @param cls the Class to check, may be null * @param cls the Class to check, may be null
* @param toClass the Class to try to assign into, returns false if null * @param toClass the Class to try to assign into, returns false if null
* @param autoboxing whether to use implicit autoboxing/unboxing between primitives and wrappers * @param autoboxing whether to use implicit autoboxing/unboxing between primitives and wrappers
* @return {@code true} if assignment possible * @return {@code true} if assignment possible
*/ */
...@@ -876,7 +892,9 @@ public final class ExtractionUtils { ...@@ -876,7 +892,9 @@ public final class ExtractionUtils {
return toClass.isAssignableFrom(cls); return toClass.isAssignableFrom(cls);
} }
/** Maps primitive {@code Class}es to their corresponding wrapper {@code Class}. */ /**
* Maps primitive {@code Class}es to their corresponding wrapper {@code Class}.
*/
private static final Map<Class<?>, Class<?>> primitiveWrapperMap = new HashMap<>(); private static final Map<Class<?>, Class<?>> primitiveWrapperMap = new HashMap<>();
static { static {
...@@ -891,7 +909,9 @@ public final class ExtractionUtils { ...@@ -891,7 +909,9 @@ public final class ExtractionUtils {
primitiveWrapperMap.put(Void.TYPE, Void.TYPE); primitiveWrapperMap.put(Void.TYPE, Void.TYPE);
} }
/** Maps wrapper {@code Class}es to their corresponding primitive types. */ /**
* Maps wrapper {@code Class}es to their corresponding primitive types.
*/
private static final Map<Class<?>, Class<?>> wrapperPrimitiveMap = new HashMap<>(); private static final Map<Class<?>, Class<?>> wrapperPrimitiveMap = new HashMap<>();
static { static {
...@@ -910,7 +930,7 @@ public final class ExtractionUtils { ...@@ -910,7 +930,7 @@ public final class ExtractionUtils {
* *
* @param cls the class to convert, may be null * @param cls the class to convert, may be null
* @return the wrapper class for {@code cls} or {@code cls} if {@code cls} is not a primitive. * @return the wrapper class for {@code cls} or {@code cls} if {@code cls} is not a primitive.
* {@code null} if null input. * {@code null} if null input.
* @since 2.1 * @since 2.1
*/ */
public static Class<?> primitiveToWrapper(final Class<?> cls) { public static Class<?> primitiveToWrapper(final Class<?> cls) {
...@@ -931,7 +951,7 @@ public final class ExtractionUtils { ...@@ -931,7 +951,7 @@ public final class ExtractionUtils {
* *
* @param cls the class to convert, may be <b>null</b> * @param cls the class to convert, may be <b>null</b>
* @return the corresponding primitive type if {@code cls} is a wrapper class, <b>null</b> * @return the corresponding primitive type if {@code cls} is a wrapper class, <b>null</b>
* otherwise * otherwise
* @see #primitiveToWrapper(Class) * @see #primitiveToWrapper(Class)
* @since 2.4 * @since 2.4
*/ */
......
dlink-client/dlink-client-1.12/src/main/java/com/dlink/cdc/FlinkCDCMergeBuilder.java
View file @ f6d81b81
...@@ -21,10 +21,10 @@ import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer; ...@@ -21,10 +21,10 @@ import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
public class FlinkCDCMergeBuilder { public class FlinkCDCMergeBuilder {
public static void buildMySqlCDC(StreamExecutionEnvironment env, FlinkCDCConfig config) { public static void buildMySqlCDC(StreamExecutionEnvironment env, FlinkCDCConfig config) {
if(Asserts.isNotNull(config.getParallelism())){ if (Asserts.isNotNull(config.getParallelism())) {
env.setParallelism(config.getParallelism()); env.setParallelism(config.getParallelism());
} }
if(Asserts.isNotNull(config.getCheckpoint())){ if (Asserts.isNotNull(config.getCheckpoint())) {
env.enableCheckpointing(config.getCheckpoint()); env.enableCheckpointing(config.getCheckpoint());
} }
MySqlSourceBuilder<String> sourceBuilder = MySqlSource.<String>builder() MySqlSourceBuilder<String> sourceBuilder = MySqlSource.<String>builder()
...@@ -32,16 +32,16 @@ public class FlinkCDCMergeBuilder { ...@@ -32,16 +32,16 @@ public class FlinkCDCMergeBuilder {
.port(config.getPort()) .port(config.getPort())
.username(config.getUsername()) .username(config.getUsername())
.password(config.getPassword()); .password(config.getPassword());
if(Asserts.isNotNull(config.getDatabase())&&config.getDatabase().size()>0){ if (Asserts.isNotNull(config.getDatabase()) && config.getDatabase().size() > 0) {
sourceBuilder.databaseList(config.getDatabase().toArray(new String[0])); sourceBuilder.databaseList(config.getDatabase().toArray(new String[0]));
} }
if(Asserts.isNotNull(config.getTable())&&config.getTable().size()>0){ if (Asserts.isNotNull(config.getTable()) && config.getTable().size() > 0) {
sourceBuilder.tableList(config.getTable().toArray(new String[0])); sourceBuilder.tableList(config.getTable().toArray(new String[0]));
} }
MySqlSourceBuilder<String> builder = sourceBuilder MySqlSourceBuilder<String> builder = sourceBuilder
.deserializer(new JsonDebeziumDeserializationSchema()); .deserializer(new JsonDebeziumDeserializationSchema());
if(Asserts.isNotNullString(config.getStartupMode())){ if (Asserts.isNotNullString(config.getStartupMode())) {
switch (config.getStartupMode().toUpperCase()){ switch (config.getStartupMode().toUpperCase()) {
case "INITIAL": case "INITIAL":
builder.startupOptions(StartupOptions.initial()); builder.startupOptions(StartupOptions.initial());
break; break;
...@@ -54,12 +54,12 @@ public class FlinkCDCMergeBuilder { ...@@ -54,12 +54,12 @@ public class FlinkCDCMergeBuilder {
default: default:
builder.startupOptions(StartupOptions.latest()); builder.startupOptions(StartupOptions.latest());
} }
}else { } else {
builder.startupOptions(StartupOptions.latest()); builder.startupOptions(StartupOptions.latest());
} }
MySqlSource<String> sourceFunction = builder.build(); MySqlSource<String> sourceFunction = builder.build();
DataStreamSource<String> streamSource = env.fromSource(sourceFunction, WatermarkStrategy.noWatermarks(), "MySQL Source"); DataStreamSource<String> streamSource = env.fromSource(sourceFunction, WatermarkStrategy.noWatermarks(), "MySQL Source");
streamSource.addSink(getKafkaProducer(config.getBrokers(),config.getTopic())); streamSource.addSink(getKafkaProducer(config.getBrokers(), config.getTopic()));
} }
private static FlinkKafkaProducer<String> getKafkaProducer(String brokers, String topic) { private static FlinkKafkaProducer<String> getKafkaProducer(String brokers, String topic) {
......
dlink-client/dlink-client-1.12/src/main/java/com/dlink/executor/CustomTableResultImpl.java
View file @ f6d81b81
...@@ -2,11 +2,7 @@ package com.dlink.executor; ...@@ -2,11 +2,7 @@ package com.dlink.executor;
import org.apache.flink.annotation.Internal; import org.apache.flink.annotation.Internal;
import org.apache.flink.core.execution.JobClient; import org.apache.flink.core.execution.JobClient;
import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.*;
import org.apache.flink.table.api.ResultKind;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.utils.PrintUtils; import org.apache.flink.table.utils.PrintUtils;
import org.apache.flink.types.Row; import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator; import org.apache.flink.util.CloseableIterator;
...@@ -18,17 +14,13 @@ import java.util.Collections; ...@@ -18,17 +14,13 @@ import java.util.Collections;
import java.util.Iterator; import java.util.Iterator;
import java.util.List; import java.util.List;
import java.util.Optional; import java.util.Optional;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/** /**
* 定制CustomTableResultImpl * 定制CustomTableResultImpl
* @author wenmo *
* @since 2021/6/7 22:06 * @author wenmo
* @since 2021/6/7 22:06
**/ **/
@Internal @Internal
class CustomTableResultImpl implements TableResult { class CustomTableResultImpl implements TableResult {
...@@ -60,12 +52,12 @@ class CustomTableResultImpl implements TableResult { ...@@ -60,12 +52,12 @@ class CustomTableResultImpl implements TableResult {
this.printStyle = Preconditions.checkNotNull(printStyle, "printStyle should not be null"); this.printStyle = Preconditions.checkNotNull(printStyle, "printStyle should not be null");
} }
public static TableResult buildTableResult(List<TableSchemaField> fields,List<Row> rows){ public static TableResult buildTableResult(List<TableSchemaField> fields, List<Row> rows) {
Builder builder = builder().resultKind(ResultKind.SUCCESS); Builder builder = builder().resultKind(ResultKind.SUCCESS);
if(fields.size()>0) { if (fields.size() > 0) {
TableSchema.Builder tableSchemaBuild = TableSchema.builder(); TableSchema.Builder tableSchemaBuild = TableSchema.builder();
for (int i = 0; i < fields.size(); i++) { for (int i = 0; i < fields.size(); i++) {
tableSchemaBuild.field(fields.get(i).getName(),fields.get(i).getType()); tableSchemaBuild.field(fields.get(i).getName(), fields.get(i).getType());
} }
builder.tableSchema(tableSchemaBuild.build()).data(rows); builder.tableSchema(tableSchemaBuild.build()).data(rows);
} }
...@@ -169,7 +161,9 @@ class CustomTableResultImpl implements TableResult { ...@@ -169,7 +161,9 @@ class CustomTableResultImpl implements TableResult {
return new Builder(); return new Builder();
} }
/** Builder for creating a {@link CustomTableResultImpl}. */ /**
* Builder for creating a {@link CustomTableResultImpl}.
*/
public static class Builder { public static class Builder {
private JobClient jobClient = null; private JobClient jobClient = null;
private TableSchema tableSchema = null; private TableSchema tableSchema = null;
...@@ -178,7 +172,8 @@ class CustomTableResultImpl implements TableResult { ...@@ -178,7 +172,8 @@ class CustomTableResultImpl implements TableResult {
private PrintStyle printStyle = private PrintStyle printStyle =
PrintStyle.tableau(Integer.MAX_VALUE, PrintUtils.NULL_COLUMN, false, false); PrintStyle.tableau(Integer.MAX_VALUE, PrintUtils.NULL_COLUMN, false, false);
private Builder() {} private Builder() {
}
/** /**
* Specifies job client which associates the submitted Flink job. * Specifies job client which associates the submitted Flink job.
...@@ -234,20 +229,26 @@ class CustomTableResultImpl implements TableResult { ...@@ -234,20 +229,26 @@ class CustomTableResultImpl implements TableResult {
return this; return this;
} }
/** Specifies print style. Default is {@link TableauStyle} with max integer column width. */ /**
* Specifies print style. Default is {@link TableauStyle} with max integer column width.
*/
public Builder setPrintStyle(PrintStyle printStyle) { public Builder setPrintStyle(PrintStyle printStyle) {
Preconditions.checkNotNull(printStyle, "printStyle should not be null"); Preconditions.checkNotNull(printStyle, "printStyle should not be null");
this.printStyle = printStyle; this.printStyle = printStyle;
return this; return this;
} }
/** Returns a {@link TableResult} instance. */ /**
* Returns a {@link TableResult} instance.
*/
public TableResult build() { public TableResult build() {
return new CustomTableResultImpl(jobClient, tableSchema, resultKind, data, printStyle); return new CustomTableResultImpl(jobClient, tableSchema, resultKind, data, printStyle);
} }
} }
/** Root interface for all print styles. */ /**
* Root interface for all print styles.
*/
public interface PrintStyle { public interface PrintStyle {
/** /**
* Create a tableau print style with given max column width, null column, change mode * Create a tableau print style with given max column width, null column, change mode
...@@ -275,7 +276,9 @@ class CustomTableResultImpl implements TableResult { ...@@ -275,7 +276,9 @@ class CustomTableResultImpl implements TableResult {
} }
} }
/** print the result schema and content as tableau form. */ /**
* print the result schema and content as tableau form.
*/
private static final class TableauStyle implements PrintStyle { private static final class TableauStyle implements PrintStyle {
/** /**
* A flag to indicate whether the column width is derived from type (true) or content * A flag to indicate whether the column width is derived from type (true) or content
...@@ -285,7 +288,9 @@ class CustomTableResultImpl implements TableResult { ...@@ -285,7 +288,9 @@ class CustomTableResultImpl implements TableResult {
private final int maxColumnWidth; private final int maxColumnWidth;
private final String nullColumn; private final String nullColumn;
/** A flag to indicate whether print row kind info. */ /**
* A flag to indicate whether print row kind info.
*/
private final boolean printRowKind; private final boolean printRowKind;
private TableauStyle( private TableauStyle(
...@@ -319,7 +324,8 @@ class CustomTableResultImpl implements TableResult { ...@@ -319,7 +324,8 @@ class CustomTableResultImpl implements TableResult {
/** /**
* only print the result content as raw form. column delimiter is ",", row delimiter is "\n". * only print the result content as raw form. column delimiter is ",", row delimiter is "\n".
*/ */
private static final class RawContentStyle implements PrintStyle {} private static final class RawContentStyle implements PrintStyle {
}
/** /**
* A {@link CloseableIterator} wrapper class that can return whether the first row is ready. * A {@link CloseableIterator} wrapper class that can return whether the first row is ready.
......
dlink-client/dlink-client-1.12/src/main/java/com/dlink/executor/TableSchemaField.java
View file @ f6d81b81
...@@ -3,8 +3,8 @@ package com.dlink.executor; ...@@ -3,8 +3,8 @@ package com.dlink.executor;
import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.DataType;
/** /**
* @author wenmo * @author wenmo
* @since 2021/6/7 22:06 * @since 2021/6/7 22:06
**/ **/
public class TableSchemaField { public class TableSchemaField {
private String name; private String name;
......
dlink-client/dlink-client-1.12/src/main/java/com/dlink/utils/FlinkUtil.java
View file @ f6d81b81
...@@ -17,18 +17,18 @@ import java.util.Optional; ...@@ -17,18 +17,18 @@ import java.util.Optional;
*/ */
public class FlinkUtil { public class FlinkUtil {
public static List<String> getFieldNamesFromCatalogManager(CatalogManager catalogManager, String catalog, String database, String table){ public static List<String> getFieldNamesFromCatalogManager(CatalogManager catalogManager, String catalog, String database, String table) {
Optional<CatalogManager.TableLookupResult> tableOpt = catalogManager.getTable( Optional<CatalogManager.TableLookupResult> tableOpt = catalogManager.getTable(
ObjectIdentifier.of(catalog, database, table) ObjectIdentifier.of(catalog, database, table)
); );
if (tableOpt.isPresent()) { if (tableOpt.isPresent()) {
return Arrays.asList(tableOpt.get().getResolvedSchema().getFieldNames()); return Arrays.asList(tableOpt.get().getResolvedSchema().getFieldNames());
}else{ } else {
return new ArrayList<String>(); return new ArrayList<String>();
} }
} }
public static List<String> catchColumn(TableResult tableResult){ public static List<String> catchColumn(TableResult tableResult) {
return Arrays.asList(tableResult.getTableSchema().getFieldNames()); return Arrays.asList(tableResult.getTableSchema().getFieldNames());
} }
} }
dlink-client/dlink-client-1.12/src/main/java/org/apache/flink/table/types/extraction/ExtractionUtils.java
View file @ f6d81b81
...@@ -21,48 +21,28 @@ package org.apache.flink.table.types.extraction; ...@@ -21,48 +21,28 @@ package org.apache.flink.table.types.extraction;
import com.dlink.pool.ClassPool; import com.dlink.pool.ClassPool;
import org.apache.flink.annotation.Internal; import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.shaded.asm7.org.objectweb.asm.*;
import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.DataTypeFactory; import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.StructuredType; import org.apache.flink.table.types.logical.StructuredType;
import org.apache.flink.shaded.asm7.org.objectweb.asm.ClassReader;
import org.apache.flink.shaded.asm7.org.objectweb.asm.ClassVisitor;
import org.apache.flink.shaded.asm7.org.objectweb.asm.Label;
import org.apache.flink.shaded.asm7.org.objectweb.asm.MethodVisitor;
import org.apache.flink.shaded.asm7.org.objectweb.asm.Opcodes;
import javax.annotation.Nullable; import javax.annotation.Nullable;
import java.io.IOException; import java.io.IOException;
import java.lang.annotation.Annotation; import java.lang.annotation.Annotation;
import java.lang.reflect.Constructor;
import java.lang.reflect.Executable;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.lang.reflect.Parameter;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type; import java.lang.reflect.Type;
import java.lang.reflect.TypeVariable; import java.lang.reflect.*;
import java.util.ArrayList; import java.util.*;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import java.util.stream.Stream; import java.util.stream.Stream;
import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getConstructorDescriptor; import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getConstructorDescriptor;
import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getMethodDescriptor; import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getMethodDescriptor;
/** Utilities for performing reflection tasks. */ /**
* Utilities for performing reflection tasks.
*/
@Internal @Internal
public final class ExtractionUtils { public final class ExtractionUtils {
...@@ -70,7 +50,9 @@ public final class ExtractionUtils { ...@@ -70,7 +50,9 @@ public final class ExtractionUtils {
// Methods shared across packages // Methods shared across packages
// -------------------------------------------------------------------------------------------- // --------------------------------------------------------------------------------------------
/** Collects methods of the given name. */ /**
* Collects methods of the given name.
*/
public static List<Method> collectMethods(Class<?> function, String methodName) { public static List<Method> collectMethods(Class<?> function, String methodName) {
return Arrays.stream(function.getMethods()) return Arrays.stream(function.getMethods())
.filter(method -> method.getName().equals(methodName)) .filter(method -> method.getName().equals(methodName))
...@@ -130,7 +112,9 @@ public final class ExtractionUtils { ...@@ -130,7 +112,9 @@ public final class ExtractionUtils {
return clz == null || ExtractionUtils.isAssignable(clz, param, true); return clz == null || ExtractionUtils.isAssignable(clz, param, true);
} }
/** Creates a method signature string like {@code int eval(Integer, String)}. */ /**
* Creates a method signature string like {@code int eval(Integer, String)}.
*/
public static String createMethodSignatureString( public static String createMethodSignatureString(
String methodName, Class<?>[] parameters, @Nullable Class<?> returnType) { String methodName, Class<?>[] parameters, @Nullable Class<?> returnType) {
final StringBuilder builder = new StringBuilder(); final StringBuilder builder = new StringBuilder();
...@@ -299,7 +283,9 @@ public final class ExtractionUtils { ...@@ -299,7 +283,9 @@ public final class ExtractionUtils {
return false; return false;
} }
/** Checks whether a field is directly readable without a getter. */ /**
* Checks whether a field is directly readable without a getter.
*/
public static boolean isStructuredFieldDirectlyReadable(Field field) { public static boolean isStructuredFieldDirectlyReadable(Field field) {
final int m = field.getModifiers(); final int m = field.getModifiers();
...@@ -307,7 +293,9 @@ public final class ExtractionUtils { ...@@ -307,7 +293,9 @@ public final class ExtractionUtils {
return Modifier.isPublic(m); return Modifier.isPublic(m);
} }
/** Checks whether a field is directly writable without a setter or constructor. */ /**
* Checks whether a field is directly writable without a setter or constructor.
*/
public static boolean isStructuredFieldDirectlyWritable(Field field) { public static boolean isStructuredFieldDirectlyWritable(Field field) {
final int m = field.getModifiers(); final int m = field.getModifiers();
...@@ -324,12 +312,16 @@ public final class ExtractionUtils { ...@@ -324,12 +312,16 @@ public final class ExtractionUtils {
// Methods intended for this package // Methods intended for this package
// -------------------------------------------------------------------------------------------- // --------------------------------------------------------------------------------------------
/** Helper method for creating consistent exceptions during extraction. */ /**
* Helper method for creating consistent exceptions during extraction.
*/
static ValidationException extractionError(String message, Object... args) { static ValidationException extractionError(String message, Object... args) {
return extractionError(null, message, args); return extractionError(null, message, args);
} }
/** Helper method for creating consistent exceptions during extraction. */ /**
* Helper method for creating consistent exceptions during extraction.
*/
static ValidationException extractionError(Throwable cause, String message, Object... args) { static ValidationException extractionError(Throwable cause, String message, Object... args) {
return new ValidationException(String.format(message, args), cause); return new ValidationException(String.format(message, args), cause);
} }
...@@ -358,8 +350,11 @@ public final class ExtractionUtils { ...@@ -358,8 +350,11 @@ public final class ExtractionUtils {
return typeHierarchy; return typeHierarchy;
} }
/** Converts a {@link Type} to {@link Class} if possible, {@code null} otherwise. */ /**
static @Nullable Class<?> toClass(Type type) { * Converts a {@link Type} to {@link Class} if possible, {@code null} otherwise.
*/
static @Nullable
Class<?> toClass(Type type) {
if (type instanceof Class) { if (type instanceof Class) {
return (Class<?>) type; return (Class<?>) type;
} else if (type instanceof ParameterizedType) { } else if (type instanceof ParameterizedType) {
...@@ -370,7 +365,9 @@ public final class ExtractionUtils { ...@@ -370,7 +365,9 @@ public final class ExtractionUtils {
return null; return null;
} }
/** Creates a raw data type. */ /**
* Creates a raw data type.
*/
@SuppressWarnings({"unchecked", "rawtypes"}) @SuppressWarnings({"unchecked", "rawtypes"})
static DataType createRawType( static DataType createRawType(
DataTypeFactory typeFactory, DataTypeFactory typeFactory,
...@@ -404,7 +401,9 @@ public final class ExtractionUtils { ...@@ -404,7 +401,9 @@ public final class ExtractionUtils {
} }
} }
/** Resolves a {@link TypeVariable} using the given type hierarchy if possible. */ /**
* Resolves a {@link TypeVariable} using the given type hierarchy if possible.
*/
static Type resolveVariable(List<Type> typeHierarchy, TypeVariable<?> variable) { static Type resolveVariable(List<Type> typeHierarchy, TypeVariable<?> variable) {
// iterate through hierarchy from top to bottom until type variable gets a non-variable // iterate through hierarchy from top to bottom until type variable gets a non-variable
// assigned // assigned
...@@ -427,7 +426,8 @@ public final class ExtractionUtils { ...@@ -427,7 +426,8 @@ public final class ExtractionUtils {
return variable; return variable;
} }
private static @Nullable Type resolveVariableInParameterizedType( private static @Nullable
Type resolveVariableInParameterizedType(
TypeVariable<?> variable, ParameterizedType currentType) { TypeVariable<?> variable, ParameterizedType currentType) {
final Class<?> currentRaw = (Class<?>) currentType.getRawType(); final Class<?> currentRaw = (Class<?>) currentType.getRawType();
final TypeVariable<?>[] currentVariables = currentRaw.getTypeParameters(); final TypeVariable<?>[] currentVariables = currentRaw.getTypeParameters();
...@@ -465,7 +465,9 @@ public final class ExtractionUtils { ...@@ -465,7 +465,9 @@ public final class ExtractionUtils {
} }
} }
/** Returns the fields of a class for a {@link StructuredType}. */ /**
* Returns the fields of a class for a {@link StructuredType}.
*/
static List<Field> collectStructuredFields(Class<?> clazz) { static List<Field> collectStructuredFields(Class<?> clazz) {
final List<Field> fields = new ArrayList<>(); final List<Field> fields = new ArrayList<>();
while (clazz != Object.class) { while (clazz != Object.class) {
...@@ -482,7 +484,9 @@ public final class ExtractionUtils { ...@@ -482,7 +484,9 @@ public final class ExtractionUtils {
return fields; return fields;
} }
/** Validates if a field is properly readable either directly or through a getter. */ /**
* Validates if a field is properly readable either directly or through a getter.
*/
static void validateStructuredFieldReadability(Class<?> clazz, Field field) { static void validateStructuredFieldReadability(Class<?> clazz, Field field) {
// field is accessible // field is accessible
if (isStructuredFieldDirectlyReadable(field)) { if (isStructuredFieldDirectlyReadable(field)) {
...@@ -525,7 +529,9 @@ public final class ExtractionUtils { ...@@ -525,7 +529,9 @@ public final class ExtractionUtils {
field.getName(), clazz.getName()); field.getName(), clazz.getName());
} }
/** Returns the boxed type of a primitive type. */ /**
* Returns the boxed type of a primitive type.
*/
static Type primitiveToWrapper(Type type) { static Type primitiveToWrapper(Type type) {
if (type instanceof Class) { if (type instanceof Class) {
return primitiveToWrapper((Class<?>) type); return primitiveToWrapper((Class<?>) type);
...@@ -533,7 +539,9 @@ public final class ExtractionUtils { ...@@ -533,7 +539,9 @@ public final class ExtractionUtils {
return type; return type;
} }
/** Collects all methods that qualify as methods of a {@link StructuredType}. */ /**
* Collects all methods that qualify as methods of a {@link StructuredType}.
*/
static List<Method> collectStructuredMethods(Class<?> clazz) { static List<Method> collectStructuredMethods(Class<?> clazz) {
final List<Method> methods = new ArrayList<>(); final List<Method> methods = new ArrayList<>();
while (clazz != Object.class) { while (clazz != Object.class) {
...@@ -584,7 +592,9 @@ public final class ExtractionUtils { ...@@ -584,7 +592,9 @@ public final class ExtractionUtils {
// Parameter Extraction Utilities // Parameter Extraction Utilities
// -------------------------------------------------------------------------------------------- // --------------------------------------------------------------------------------------------
/** Result of the extraction in {@link #extractAssigningConstructor(Class, List)}. */ /**
* Result of the extraction in {@link #extractAssigningConstructor(Class, List)}.
*/
static class AssigningConstructor { static class AssigningConstructor {
public final Constructor<?> constructor; public final Constructor<?> constructor;
public final List<String> parameterNames; public final List<String> parameterNames;
...@@ -599,7 +609,8 @@ public final class ExtractionUtils { ...@@ -599,7 +609,8 @@ public final class ExtractionUtils {
* Checks whether the given constructor takes all of the given fields with matching (possibly * Checks whether the given constructor takes all of the given fields with matching (possibly
* primitive) type and name. An assigning constructor can define the order of fields. * primitive) type and name. An assigning constructor can define the order of fields.
*/ */
static @Nullable AssigningConstructor extractAssigningConstructor( static @Nullable
AssigningConstructor extractAssigningConstructor(
Class<?> clazz, List<Field> fields) { Class<?> clazz, List<Field> fields) {
AssigningConstructor foundConstructor = null; AssigningConstructor foundConstructor = null;
for (Constructor<?> constructor : clazz.getDeclaredConstructors()) { for (Constructor<?> constructor : clazz.getDeclaredConstructors()) {
...@@ -623,8 +634,11 @@ public final class ExtractionUtils { ...@@ -623,8 +634,11 @@ public final class ExtractionUtils {
return foundConstructor; return foundConstructor;
} }
/** Extracts the parameter names of a method if possible. */ /**
static @Nullable List<String> extractMethodParameterNames(Method method) { * Extracts the parameter names of a method if possible.
*/
static @Nullable
List<String> extractMethodParameterNames(Method method) {
return extractExecutableNames(method); return extractExecutableNames(method);
} }
...@@ -632,7 +646,8 @@ public final class ExtractionUtils { ...@@ -632,7 +646,8 @@ public final class ExtractionUtils {
* Extracts ordered parameter names from a constructor that takes all of the given fields with * Extracts ordered parameter names from a constructor that takes all of the given fields with
* matching (possibly primitive) type and name. * matching (possibly primitive) type and name.
*/ */
private static @Nullable List<String> extractConstructorParameterNames( private static @Nullable
List<String> extractConstructorParameterNames(
Constructor<?> constructor, List<Field> fields) { Constructor<?> constructor, List<Field> fields) {
final Type[] parameterTypes = constructor.getGenericParameterTypes(); final Type[] parameterTypes = constructor.getGenericParameterTypes();
...@@ -659,7 +674,8 @@ public final class ExtractionUtils { ...@@ -659,7 +674,8 @@ public final class ExtractionUtils {
return parameterNames; return parameterNames;
} }
private static @Nullable List<String> extractExecutableNames(Executable executable) { private static @Nullable
List<String> extractExecutableNames(Executable executable) {
final int offset; final int offset;
if (!Modifier.isStatic(executable.getModifiers())) { if (!Modifier.isStatic(executable.getModifiers())) {
// remove "this" as first parameter // remove "this" as first parameter
...@@ -705,7 +721,7 @@ public final class ExtractionUtils { ...@@ -705,7 +721,7 @@ public final class ExtractionUtils {
private static ClassReader getClassReader(Class<?> cls) { private static ClassReader getClassReader(Class<?> cls) {
final String className = cls.getName().replaceFirst("^.*\\.", "") + ".class"; final String className = cls.getName().replaceFirst("^.*\\.", "") + ".class";
if(ClassPool.exist(cls.getName())){ if (ClassPool.exist(cls.getName())) {
return new ClassReader(ClassPool.get(cls.getName()).getClassByte()); return new ClassReader(ClassPool.get(cls.getName()).getClassByte());
} }
try { try {
...@@ -798,8 +814,8 @@ public final class ExtractionUtils { ...@@ -798,8 +814,8 @@ public final class ExtractionUtils {
* href="http://docs.oracle.com/javase/specs/">The Java Language Specification</a></em>, * href="http://docs.oracle.com/javase/specs/">The Java Language Specification</a></em>,
* sections 5.1.1, 5.1.2 and 5.1.4 for details. * sections 5.1.1, 5.1.2 and 5.1.4 for details.
* *
* @param cls the Class to check, may be null * @param cls the Class to check, may be null
* @param toClass the Class to try to assign into, returns false if null * @param toClass the Class to try to assign into, returns false if null
* @param autoboxing whether to use implicit autoboxing/unboxing between primitives and wrappers * @param autoboxing whether to use implicit autoboxing/unboxing between primitives and wrappers
* @return {@code true} if assignment possible * @return {@code true} if assignment possible
*/ */
...@@ -876,7 +892,9 @@ public final class ExtractionUtils { ...@@ -876,7 +892,9 @@ public final class ExtractionUtils {
return toClass.isAssignableFrom(cls); return toClass.isAssignableFrom(cls);
} }
/** Maps primitive {@code Class}es to their corresponding wrapper {@code Class}. */ /**
* Maps primitive {@code Class}es to their corresponding wrapper {@code Class}.
*/
private static final Map<Class<?>, Class<?>> primitiveWrapperMap = new HashMap<>(); private static final Map<Class<?>, Class<?>> primitiveWrapperMap = new HashMap<>();
static { static {
...@@ -891,7 +909,9 @@ public final class ExtractionUtils { ...@@ -891,7 +909,9 @@ public final class ExtractionUtils {
primitiveWrapperMap.put(Void.TYPE, Void.TYPE); primitiveWrapperMap.put(Void.TYPE, Void.TYPE);
} }
/** Maps wrapper {@code Class}es to their corresponding primitive types. */ /**
* Maps wrapper {@code Class}es to their corresponding primitive types.
*/
private static final Map<Class<?>, Class<?>> wrapperPrimitiveMap = new HashMap<>(); private static final Map<Class<?>, Class<?>> wrapperPrimitiveMap = new HashMap<>();
static { static {
...@@ -910,7 +930,7 @@ public final class ExtractionUtils { ...@@ -910,7 +930,7 @@ public final class ExtractionUtils {
* *
* @param cls the class to convert, may be null * @param cls the class to convert, may be null
* @return the wrapper class for {@code cls} or {@code cls} if {@code cls} is not a primitive. * @return the wrapper class for {@code cls} or {@code cls} if {@code cls} is not a primitive.
* {@code null} if null input. * {@code null} if null input.
* @since 2.1 * @since 2.1
*/ */
public static Class<?> primitiveToWrapper(final Class<?> cls) { public static Class<?> primitiveToWrapper(final Class<?> cls) {
...@@ -931,7 +951,7 @@ public final class ExtractionUtils { ...@@ -931,7 +951,7 @@ public final class ExtractionUtils {
* *
* @param cls the class to convert, may be <b>null</b> * @param cls the class to convert, may be <b>null</b>
* @return the corresponding primitive type if {@code cls} is a wrapper class, <b>null</b> * @return the corresponding primitive type if {@code cls} is a wrapper class, <b>null</b>
* otherwise * otherwise
* @see #primitiveToWrapper(Class) * @see #primitiveToWrapper(Class)
* @since 2.4 * @since 2.4
*/ */
......
dlink-client/dlink-client-1.13/src/main/java/com/dlink/cdc/FlinkCDCMergeBuilder.java
View file @ f6d81b81
...@@ -21,10 +21,10 @@ import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer; ...@@ -21,10 +21,10 @@ import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
public class FlinkCDCMergeBuilder { public class FlinkCDCMergeBuilder {
public static void buildMySqlCDC(StreamExecutionEnvironment env, FlinkCDCConfig config) { public static void buildMySqlCDC(StreamExecutionEnvironment env, FlinkCDCConfig config) {
if(Asserts.isNotNull(config.getParallelism())){ if (Asserts.isNotNull(config.getParallelism())) {
env.setParallelism(config.getParallelism()); env.setParallelism(config.getParallelism());
} }
if(Asserts.isNotNull(config.getCheckpoint())){ if (Asserts.isNotNull(config.getCheckpoint())) {
env.enableCheckpointing(config.getCheckpoint()); env.enableCheckpointing(config.getCheckpoint());
} }
MySqlSourceBuilder<String> sourceBuilder = MySqlSource.<String>builder() MySqlSourceBuilder<String> sourceBuilder = MySqlSource.<String>builder()
...@@ -32,16 +32,16 @@ public class FlinkCDCMergeBuilder { ...@@ -32,16 +32,16 @@ public class FlinkCDCMergeBuilder {
.port(config.getPort()) .port(config.getPort())
.username(config.getUsername()) .username(config.getUsername())
.password(config.getPassword()); .password(config.getPassword());
if(Asserts.isNotNull(config.getDatabase())&&config.getDatabase().size()>0){ if (Asserts.isNotNull(config.getDatabase()) && config.getDatabase().size() > 0) {
sourceBuilder.databaseList(config.getDatabase().toArray(new String[0])); sourceBuilder.databaseList(config.getDatabase().toArray(new String[0]));
} }
if(Asserts.isNotNull(config.getTable())&&config.getTable().size()>0){ if (Asserts.isNotNull(config.getTable()) && config.getTable().size() > 0) {
sourceBuilder.tableList(config.getTable().toArray(new String[0])); sourceBuilder.tableList(config.getTable().toArray(new String[0]));
} }
MySqlSourceBuilder<String> builder = sourceBuilder MySqlSourceBuilder<String> builder = sourceBuilder
.deserializer(new JsonDebeziumDeserializationSchema()); .deserializer(new JsonDebeziumDeserializationSchema());
if(Asserts.isNotNullString(config.getStartupMode())){ if (Asserts.isNotNullString(config.getStartupMode())) {
switch (config.getStartupMode().toUpperCase()){ switch (config.getStartupMode().toUpperCase()) {
case "INITIAL": case "INITIAL":
builder.startupOptions(StartupOptions.initial()); builder.startupOptions(StartupOptions.initial());
break; break;
...@@ -54,12 +54,12 @@ public class FlinkCDCMergeBuilder { ...@@ -54,12 +54,12 @@ public class FlinkCDCMergeBuilder {
default: default:
builder.startupOptions(StartupOptions.latest()); builder.startupOptions(StartupOptions.latest());
} }
}else { } else {
builder.startupOptions(StartupOptions.latest()); builder.startupOptions(StartupOptions.latest());
} }
MySqlSource<String> sourceFunction = builder.build(); MySqlSource<String> sourceFunction = builder.build();
DataStreamSource<String> streamSource = env.fromSource(sourceFunction, WatermarkStrategy.noWatermarks(), "MySQL Source"); DataStreamSource<String> streamSource = env.fromSource(sourceFunction, WatermarkStrategy.noWatermarks(), "MySQL Source");
streamSource.addSink(getKafkaProducer(config.getBrokers(),config.getTopic())); streamSource.addSink(getKafkaProducer(config.getBrokers(), config.getTopic()));
} }
private static FlinkKafkaProducer<String> getKafkaProducer(String brokers, String topic) { private static FlinkKafkaProducer<String> getKafkaProducer(String brokers, String topic) {
......
dlink-client/dlink-client-1.13/src/main/java/com/dlink/executor/CustomTableEnvironmentImpl.java
View file @ f6d81b81
...@@ -235,7 +235,7 @@ public class CustomTableEnvironmentImpl extends TableEnvironmentImpl implements ...@@ -235,7 +235,7 @@ public class CustomTableEnvironmentImpl extends TableEnvironmentImpl implements
if (setOperation.getKey().isPresent() && setOperation.getValue().isPresent()) { if (setOperation.getKey().isPresent() && setOperation.getValue().isPresent()) {
String key = setOperation.getKey().get().trim(); String key = setOperation.getKey().get().trim();
String value = setOperation.getValue().get().trim(); String value = setOperation.getValue().get().trim();
if(Asserts.isNullString(key)||Asserts.isNullString(value)){ if (Asserts.isNullString(key) || Asserts.isNullString(value)) {
return; return;
} }
Map<String, String> confMap = new HashMap<>(); Map<String, String> confMap = new HashMap<>();
...@@ -250,7 +250,7 @@ public class CustomTableEnvironmentImpl extends TableEnvironmentImpl implements ...@@ -250,7 +250,7 @@ public class CustomTableEnvironmentImpl extends TableEnvironmentImpl implements
private void callReset(ResetOperation resetOperation, StreamExecutionEnvironment environment, Map<String, Object> setMap) { private void callReset(ResetOperation resetOperation, StreamExecutionEnvironment environment, Map<String, Object> setMap) {
if (resetOperation.getKey().isPresent()) { if (resetOperation.getKey().isPresent()) {
String key = resetOperation.getKey().get().trim(); String key = resetOperation.getKey().get().trim();
if(Asserts.isNullString(key)){ if (Asserts.isNullString(key)) {
return; return;
} }
Map<String, String> confMap = new HashMap<>(); Map<String, String> confMap = new HashMap<>();
......
dlink-client/dlink-client-1.13/src/main/java/com/dlink/executor/CustomTableResultImpl.java
View file @ f6d81b81
...@@ -17,22 +17,14 @@ import org.apache.flink.util.Preconditions; ...@@ -17,22 +17,14 @@ import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable; import javax.annotation.Nullable;
import java.io.PrintWriter; import java.io.PrintWriter;
import java.time.ZoneId; import java.time.ZoneId;
import java.util.ArrayList; import java.util.*;
import java.util.Collections; import java.util.concurrent.*;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/** /**
* 定制CustomTableResultImpl * 定制CustomTableResultImpl
* @author wenmo *
* @since 2021/6/7 22:06 * @author wenmo
* @since 2021/6/7 22:06
**/ **/
@Internal @Internal
public class CustomTableResultImpl implements TableResult { public class CustomTableResultImpl implements TableResult {
...@@ -68,16 +60,16 @@ public class CustomTableResultImpl implements TableResult { ...@@ -68,16 +60,16 @@ public class CustomTableResultImpl implements TableResult {
Preconditions.checkNotNull(sessionTimeZone, "sessionTimeZone should not be null"); Preconditions.checkNotNull(sessionTimeZone, "sessionTimeZone should not be null");
} }
public static TableResult buildTableResult(List<TableSchemaField> fields,List<Row> rows){ public static TableResult buildTableResult(List<TableSchemaField> fields, List<Row> rows) {
Builder builder = builder().resultKind(ResultKind.SUCCESS); Builder builder = builder().resultKind(ResultKind.SUCCESS);
if(fields.size()>0) { if (fields.size() > 0) {
List<String> columnNames = new ArrayList<>(); List<String> columnNames = new ArrayList<>();
List<DataType> columnTypes = new ArrayList<>(); List<DataType> columnTypes = new ArrayList<>();
for (int i = 0; i < fields.size(); i++) { for (int i = 0; i < fields.size(); i++) {
columnNames.add(fields.get(i).getName()); columnNames.add(fields.get(i).getName());
columnTypes.add(fields.get(i).getType()); columnTypes.add(fields.get(i).getType());
} }
builder.schema(ResolvedSchema.physical(columnNames,columnTypes)).data(rows); builder.schema(ResolvedSchema.physical(columnNames, columnTypes)).data(rows);
} }
return builder.build(); return builder.build();
} }
...@@ -184,7 +176,9 @@ public class CustomTableResultImpl implements TableResult { ...@@ -184,7 +176,9 @@ public class CustomTableResultImpl implements TableResult {
return new Builder(); return new Builder();
} }
/** Builder for creating a {@link CustomTableResultImpl}. */ /**
* Builder for creating a {@link CustomTableResultImpl}.
*/
public static class Builder { public static class Builder {
private JobClient jobClient = null; private JobClient jobClient = null;
private ResolvedSchema resolvedSchema = null; private ResolvedSchema resolvedSchema = null;
...@@ -194,7 +188,8 @@ public class CustomTableResultImpl implements TableResult { ...@@ -194,7 +188,8 @@ public class CustomTableResultImpl implements TableResult {
PrintStyle.tableau(Integer.MAX_VALUE, PrintUtils.NULL_COLUMN, false, false); PrintStyle.tableau(Integer.MAX_VALUE, PrintUtils.NULL_COLUMN, false, false);
private ZoneId sessionTimeZone = ZoneId.of("UTC"); private ZoneId sessionTimeZone = ZoneId.of("UTC");
private Builder() {} private Builder() {
}
/** /**
* Specifies job client which associates the submitted Flink job. * Specifies job client which associates the submitted Flink job.
...@@ -250,28 +245,36 @@ public class CustomTableResultImpl implements TableResult { ...@@ -250,28 +245,36 @@ public class CustomTableResultImpl implements TableResult {
return this; return this;
} }
/** Specifies print style. Default is {@link TableauStyle} with max integer column width. */ /**
* Specifies print style. Default is {@link TableauStyle} with max integer column width.
*/
public Builder setPrintStyle(PrintStyle printStyle) { public Builder setPrintStyle(PrintStyle printStyle) {
Preconditions.checkNotNull(printStyle, "printStyle should not be null"); Preconditions.checkNotNull(printStyle, "printStyle should not be null");
this.printStyle = printStyle; this.printStyle = printStyle;
return this; return this;
} }
/** Specifies session time zone. */ /**
* Specifies session time zone.
*/
public Builder setSessionTimeZone(ZoneId sessionTimeZone) { public Builder setSessionTimeZone(ZoneId sessionTimeZone) {
Preconditions.checkNotNull(sessionTimeZone, "sessionTimeZone should not be null"); Preconditions.checkNotNull(sessionTimeZone, "sessionTimeZone should not be null");
this.sessionTimeZone = sessionTimeZone; this.sessionTimeZone = sessionTimeZone;
return this; return this;
} }
/** Returns a {@link TableResult} instance. */ /**
* Returns a {@link TableResult} instance.
*/
public TableResult build() { public TableResult build() {
return new CustomTableResultImpl( return new CustomTableResultImpl(
jobClient, resolvedSchema, resultKind, data, printStyle, sessionTimeZone); jobClient, resolvedSchema, resultKind, data, printStyle, sessionTimeZone);
} }
} }
/** Root interface for all print styles. */ /**
* Root interface for all print styles.
*/
public interface PrintStyle { public interface PrintStyle {
/** /**
* Create a tableau print style with given max column width, null column, change mode * Create a tableau print style with given max column width, null column, change mode
...@@ -299,7 +302,9 @@ public class CustomTableResultImpl implements TableResult { ...@@ -299,7 +302,9 @@ public class CustomTableResultImpl implements TableResult {
} }
} }
/** print the result schema and content as tableau form. */ /**
* print the result schema and content as tableau form.
*/
private static final class TableauStyle implements PrintStyle { private static final class TableauStyle implements PrintStyle {
/** /**
* A flag to indicate whether the column width is derived from type (true) or content * A flag to indicate whether the column width is derived from type (true) or content
...@@ -309,7 +314,9 @@ public class CustomTableResultImpl implements TableResult { ...@@ -309,7 +314,9 @@ public class CustomTableResultImpl implements TableResult {
private final int maxColumnWidth; private final int maxColumnWidth;
private final String nullColumn; private final String nullColumn;
/** A flag to indicate whether print row kind info. */ /**
* A flag to indicate whether print row kind info.
*/
private final boolean printRowKind; private final boolean printRowKind;
private TableauStyle( private TableauStyle(
...@@ -343,7 +350,8 @@ public class CustomTableResultImpl implements TableResult { ...@@ -343,7 +350,8 @@ public class CustomTableResultImpl implements TableResult {
/** /**
* only print the result content as raw form. column delimiter is ",", row delimiter is "\n". * only print the result content as raw form. column delimiter is ",", row delimiter is "\n".
*/ */
private static final class RawContentStyle implements PrintStyle {} private static final class RawContentStyle implements PrintStyle {
}
/** /**
* A {@link CloseableIterator} wrapper class that can return whether the first row is ready. * A {@link CloseableIterator} wrapper class that can return whether the first row is ready.
......
dlink-client/dlink-client-1.13/src/main/java/com/dlink/executor/TableSchemaField.java
View file @ f6d81b81
...@@ -3,8 +3,8 @@ package com.dlink.executor; ...@@ -3,8 +3,8 @@ package com.dlink.executor;
import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.DataType;
/** /**
* @author wenmo * @author wenmo
* @since 2021/6/7 22:06 * @since 2021/6/7 22:06
**/ **/
public class TableSchemaField { public class TableSchemaField {
private String name; private String name;
......
dlink-client/dlink-client-1.13/src/main/java/com/dlink/utils/FlinkUtil.java
View file @ f6d81b81
...@@ -4,7 +4,9 @@ import org.apache.flink.table.api.TableResult; ...@@ -4,7 +4,9 @@ import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.catalog.CatalogManager; import org.apache.flink.table.catalog.CatalogManager;
import org.apache.flink.table.catalog.ObjectIdentifier; import org.apache.flink.table.catalog.ObjectIdentifier;
import java.util.*; import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
/** /**
* FlinkUtil * FlinkUtil
...@@ -14,18 +16,18 @@ import java.util.*; ...@@ -14,18 +16,18 @@ import java.util.*;
*/ */
public class FlinkUtil { public class FlinkUtil {
public static List<String> getFieldNamesFromCatalogManager(CatalogManager catalogManager, String catalog, String database, String table){ public static List<String> getFieldNamesFromCatalogManager(CatalogManager catalogManager, String catalog, String database, String table) {
Optional<CatalogManager.TableLookupResult> tableOpt = catalogManager.getTable( Optional<CatalogManager.TableLookupResult> tableOpt = catalogManager.getTable(
ObjectIdentifier.of(catalog, database, table) ObjectIdentifier.of(catalog, database, table)
); );
if (tableOpt.isPresent()) { if (tableOpt.isPresent()) {
return tableOpt.get().getResolvedSchema().getColumnNames(); return tableOpt.get().getResolvedSchema().getColumnNames();
}else{ } else {
return new ArrayList<>(); return new ArrayList<>();
} }
} }
public static List<String> catchColumn(TableResult tableResult){ public static List<String> catchColumn(TableResult tableResult) {
return tableResult.getResolvedSchema().getColumnNames(); return tableResult.getResolvedSchema().getColumnNames();
} }
} }
dlink-client/dlink-client-1.13/src/main/java/org/apache/flink/table/types/extraction/ExtractionUtils.java
View file @ f6d81b81
...@@ -21,41 +21,19 @@ package org.apache.flink.table.types.extraction; ...@@ -21,41 +21,19 @@ package org.apache.flink.table.types.extraction;
import com.dlink.pool.ClassPool; import com.dlink.pool.ClassPool;
import org.apache.flink.annotation.Internal; import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.shaded.asm7.org.objectweb.asm.*;
import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.DataTypeFactory; import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.StructuredType; import org.apache.flink.table.types.logical.StructuredType;
import org.apache.flink.shaded.asm7.org.objectweb.asm.ClassReader;
import org.apache.flink.shaded.asm7.org.objectweb.asm.ClassVisitor;
import org.apache.flink.shaded.asm7.org.objectweb.asm.Label;
import org.apache.flink.shaded.asm7.org.objectweb.asm.MethodVisitor;
import org.apache.flink.shaded.asm7.org.objectweb.asm.Opcodes;
import javax.annotation.Nullable; import javax.annotation.Nullable;
import java.io.IOException; import java.io.IOException;
import java.lang.annotation.Annotation; import java.lang.annotation.Annotation;
import java.lang.reflect.Constructor;
import java.lang.reflect.Executable;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.lang.reflect.Parameter;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type; import java.lang.reflect.Type;
import java.lang.reflect.TypeVariable; import java.lang.reflect.*;
import java.util.ArrayList; import java.util.*;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.function.Function; import java.util.function.Function;
import java.util.regex.Pattern; import java.util.regex.Pattern;
import java.util.stream.Collectors; import java.util.stream.Collectors;
...@@ -64,7 +42,9 @@ import java.util.stream.Stream; ...@@ -64,7 +42,9 @@ import java.util.stream.Stream;
import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getConstructorDescriptor; import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getConstructorDescriptor;
import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getMethodDescriptor; import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getMethodDescriptor;
/** Utilities for performing reflection tasks. */ /**
* Utilities for performing reflection tasks.
*/
@Internal @Internal
public final class ExtractionUtils { public final class ExtractionUtils {
...@@ -72,7 +52,9 @@ public final class ExtractionUtils { ...@@ -72,7 +52,9 @@ public final class ExtractionUtils {
// Methods shared across packages // Methods shared across packages
// -------------------------------------------------------------------------------------------- // --------------------------------------------------------------------------------------------
/** Collects methods of the given name. */ /**
* Collects methods of the given name.
*/
public static List<Method> collectMethods(Class<?> function, String methodName) { public static List<Method> collectMethods(Class<?> function, String methodName) {
return Arrays.stream(function.getMethods()) return Arrays.stream(function.getMethods())
.filter(method -> method.getName().equals(methodName)) .filter(method -> method.getName().equals(methodName))
...@@ -132,7 +114,9 @@ public final class ExtractionUtils { ...@@ -132,7 +114,9 @@ public final class ExtractionUtils {
return clz == null || ExtractionUtils.isAssignable(clz, param, true); return clz == null || ExtractionUtils.isAssignable(clz, param, true);
} }
/** Creates a method signature string like {@code int eval(Integer, String)}. */ /**
* Creates a method signature string like {@code int eval(Integer, String)}.
*/
public static String createMethodSignatureString( public static String createMethodSignatureString(
String methodName, Class<?>[] parameters, @Nullable Class<?> returnType) { String methodName, Class<?>[] parameters, @Nullable Class<?> returnType) {
final StringBuilder builder = new StringBuilder(); final StringBuilder builder = new StringBuilder();
...@@ -305,7 +289,9 @@ public final class ExtractionUtils { ...@@ -305,7 +289,9 @@ public final class ExtractionUtils {
return false; return false;
} }
/** Checks whether a field is directly readable without a getter. */ /**
* Checks whether a field is directly readable without a getter.
*/
public static boolean isStructuredFieldDirectlyReadable(Field field) { public static boolean isStructuredFieldDirectlyReadable(Field field) {
final int m = field.getModifiers(); final int m = field.getModifiers();
...@@ -313,7 +299,9 @@ public final class ExtractionUtils { ...@@ -313,7 +299,9 @@ public final class ExtractionUtils {
return Modifier.isPublic(m); return Modifier.isPublic(m);
} }
/** Checks whether a field is directly writable without a setter or constructor. */ /**
* Checks whether a field is directly writable without a setter or constructor.
*/
public static boolean isStructuredFieldDirectlyWritable(Field field) { public static boolean isStructuredFieldDirectlyWritable(Field field) {
final int m = field.getModifiers(); final int m = field.getModifiers();
...@@ -352,12 +340,16 @@ public final class ExtractionUtils { ...@@ -352,12 +340,16 @@ public final class ExtractionUtils {
// Methods intended for this package // Methods intended for this package
// -------------------------------------------------------------------------------------------- // --------------------------------------------------------------------------------------------
/** Helper method for creating consistent exceptions during extraction. */ /**
* Helper method for creating consistent exceptions during extraction.
*/
static ValidationException extractionError(String message, Object... args) { static ValidationException extractionError(String message, Object... args) {
return extractionError(null, message, args); return extractionError(null, message, args);
} }
/** Helper method for creating consistent exceptions during extraction. */ /**
* Helper method for creating consistent exceptions during extraction.
*/
static ValidationException extractionError(Throwable cause, String message, Object... args) { static ValidationException extractionError(Throwable cause, String message, Object... args) {
return new ValidationException(String.format(message, args), cause); return new ValidationException(String.format(message, args), cause);
} }
...@@ -386,8 +378,11 @@ public final class ExtractionUtils { ...@@ -386,8 +378,11 @@ public final class ExtractionUtils {
return typeHierarchy; return typeHierarchy;
} }
/** Converts a {@link Type} to {@link Class} if possible, {@code null} otherwise. */ /**
static @Nullable Class<?> toClass(Type type) { * Converts a {@link Type} to {@link Class} if possible, {@code null} otherwise.
*/
static @Nullable
Class<?> toClass(Type type) {
if (type instanceof Class) { if (type instanceof Class) {
return (Class<?>) type; return (Class<?>) type;
} else if (type instanceof ParameterizedType) { } else if (type instanceof ParameterizedType) {
...@@ -398,7 +393,9 @@ public final class ExtractionUtils { ...@@ -398,7 +393,9 @@ public final class ExtractionUtils {
return null; return null;
} }
/** Creates a raw data type. */ /**
* Creates a raw data type.
*/
@SuppressWarnings({"unchecked", "rawtypes"}) @SuppressWarnings({"unchecked", "rawtypes"})
static DataType createRawType( static DataType createRawType(
DataTypeFactory typeFactory, DataTypeFactory typeFactory,
...@@ -432,7 +429,9 @@ public final class ExtractionUtils { ...@@ -432,7 +429,9 @@ public final class ExtractionUtils {
} }
} }
/** Resolves a {@link TypeVariable} using the given type hierarchy if possible. */ /**
* Resolves a {@link TypeVariable} using the given type hierarchy if possible.
*/
static Type resolveVariable(List<Type> typeHierarchy, TypeVariable<?> variable) { static Type resolveVariable(List<Type> typeHierarchy, TypeVariable<?> variable) {
// iterate through hierarchy from top to bottom until type variable gets a non-variable // iterate through hierarchy from top to bottom until type variable gets a non-variable
// assigned // assigned
...@@ -455,7 +454,8 @@ public final class ExtractionUtils { ...@@ -455,7 +454,8 @@ public final class ExtractionUtils {
return variable; return variable;
} }
private static @Nullable Type resolveVariableInParameterizedType( private static @Nullable
Type resolveVariableInParameterizedType(
TypeVariable<?> variable, ParameterizedType currentType) { TypeVariable<?> variable, ParameterizedType currentType) {
final Class<?> currentRaw = (Class<?>) currentType.getRawType(); final Class<?> currentRaw = (Class<?>) currentType.getRawType();
final TypeVariable<?>[] currentVariables = currentRaw.getTypeParameters(); final TypeVariable<?>[] currentVariables = currentRaw.getTypeParameters();
...@@ -493,7 +493,9 @@ public final class ExtractionUtils { ...@@ -493,7 +493,9 @@ public final class ExtractionUtils {
} }
} }
/** Returns the fields of a class for a {@link StructuredType}. */ /**
* Returns the fields of a class for a {@link StructuredType}.
*/
static List<Field> collectStructuredFields(Class<?> clazz) { static List<Field> collectStructuredFields(Class<?> clazz) {
final List<Field> fields = new ArrayList<>(); final List<Field> fields = new ArrayList<>();
while (clazz != Object.class) { while (clazz != Object.class) {
...@@ -510,7 +512,9 @@ public final class ExtractionUtils { ...@@ -510,7 +512,9 @@ public final class ExtractionUtils {
return fields; return fields;
} }
/** Validates if a field is properly readable either directly or through a getter. */ /**
* Validates if a field is properly readable either directly or through a getter.
*/
static void validateStructuredFieldReadability(Class<?> clazz, Field field) { static void validateStructuredFieldReadability(Class<?> clazz, Field field) {
// field is accessible // field is accessible
if (isStructuredFieldDirectlyReadable(field)) { if (isStructuredFieldDirectlyReadable(field)) {
...@@ -553,7 +557,9 @@ public final class ExtractionUtils { ...@@ -553,7 +557,9 @@ public final class ExtractionUtils {
field.getName(), clazz.getName()); field.getName(), clazz.getName());
} }
/** Returns the boxed type of a primitive type. */ /**
* Returns the boxed type of a primitive type.
*/
static Type primitiveToWrapper(Type type) { static Type primitiveToWrapper(Type type) {
if (type instanceof Class) { if (type instanceof Class) {
return primitiveToWrapper((Class<?>) type); return primitiveToWrapper((Class<?>) type);
...@@ -561,7 +567,9 @@ public final class ExtractionUtils { ...@@ -561,7 +567,9 @@ public final class ExtractionUtils {
return type; return type;
} }
/** Collects all methods that qualify as methods of a {@link StructuredType}. */ /**
* Collects all methods that qualify as methods of a {@link StructuredType}.
*/
static List<Method> collectStructuredMethods(Class<?> clazz) { static List<Method> collectStructuredMethods(Class<?> clazz) {
final List<Method> methods = new ArrayList<>(); final List<Method> methods = new ArrayList<>();
while (clazz != Object.class) { while (clazz != Object.class) {
...@@ -612,7 +620,9 @@ public final class ExtractionUtils { ...@@ -612,7 +620,9 @@ public final class ExtractionUtils {
// Parameter Extraction Utilities // Parameter Extraction Utilities
// -------------------------------------------------------------------------------------------- // --------------------------------------------------------------------------------------------
/** Result of the extraction in {@link #extractAssigningConstructor(Class, List)}. */ /**
* Result of the extraction in {@link #extractAssigningConstructor(Class, List)}.
*/
public static class AssigningConstructor { public static class AssigningConstructor {
public final Constructor<?> constructor; public final Constructor<?> constructor;
public final List<String> parameterNames; public final List<String> parameterNames;
...@@ -627,7 +637,8 @@ public final class ExtractionUtils { ...@@ -627,7 +637,8 @@ public final class ExtractionUtils {
* Checks whether the given constructor takes all of the given fields with matching (possibly * Checks whether the given constructor takes all of the given fields with matching (possibly
* primitive) type and name. An assigning constructor can define the order of fields. * primitive) type and name. An assigning constructor can define the order of fields.
*/ */
public static @Nullable AssigningConstructor extractAssigningConstructor( public static @Nullable
AssigningConstructor extractAssigningConstructor(
Class<?> clazz, List<Field> fields) { Class<?> clazz, List<Field> fields) {
AssigningConstructor foundConstructor = null; AssigningConstructor foundConstructor = null;
for (Constructor<?> constructor : clazz.getDeclaredConstructors()) { for (Constructor<?> constructor : clazz.getDeclaredConstructors()) {
...@@ -651,8 +662,11 @@ public final class ExtractionUtils { ...@@ -651,8 +662,11 @@ public final class ExtractionUtils {
return foundConstructor; return foundConstructor;
} }
/** Extracts the parameter names of a method if possible. */ /**
static @Nullable List<String> extractMethodParameterNames(Method method) { * Extracts the parameter names of a method if possible.
*/
static @Nullable
List<String> extractMethodParameterNames(Method method) {
return extractExecutableNames(method); return extractExecutableNames(method);
} }
...@@ -660,7 +674,8 @@ public final class ExtractionUtils { ...@@ -660,7 +674,8 @@ public final class ExtractionUtils {
* Extracts ordered parameter names from a constructor that takes all of the given fields with * Extracts ordered parameter names from a constructor that takes all of the given fields with
* matching (possibly primitive and lenient) type and name. * matching (possibly primitive and lenient) type and name.
*/ */
private static @Nullable List<String> extractConstructorParameterNames( private static @Nullable
List<String> extractConstructorParameterNames(
Constructor<?> constructor, List<Field> fields) { Constructor<?> constructor, List<Field> fields) {
final Type[] parameterTypes = constructor.getGenericParameterTypes(); final Type[] parameterTypes = constructor.getGenericParameterTypes();
...@@ -697,7 +712,8 @@ public final class ExtractionUtils { ...@@ -697,7 +712,8 @@ public final class ExtractionUtils {
return fieldNames; return fieldNames;
} }
private static @Nullable List<String> extractExecutableNames(Executable executable) { private static @Nullable
List<String> extractExecutableNames(Executable executable) {
final int offset; final int offset;
if (!Modifier.isStatic(executable.getModifiers())) { if (!Modifier.isStatic(executable.getModifiers())) {
// remove "this" as first parameter // remove "this" as first parameter
...@@ -743,7 +759,7 @@ public final class ExtractionUtils { ...@@ -743,7 +759,7 @@ public final class ExtractionUtils {
private static ClassReader getClassReader(Class<?> cls) { private static ClassReader getClassReader(Class<?> cls) {
final String className = cls.getName().replaceFirst("^.*\\.", "") + ".class"; final String className = cls.getName().replaceFirst("^.*\\.", "") + ".class";
if(ClassPool.exist(cls.getName())){ if (ClassPool.exist(cls.getName())) {
return new ClassReader(ClassPool.get(cls.getName()).getClassByte()); return new ClassReader(ClassPool.get(cls.getName()).getClassByte());
} }
try { try {
...@@ -836,8 +852,8 @@ public final class ExtractionUtils { ...@@ -836,8 +852,8 @@ public final class ExtractionUtils {
* href="http://docs.oracle.com/javase/specs/">The Java Language Specification</a></em>, * href="http://docs.oracle.com/javase/specs/">The Java Language Specification</a></em>,
* sections 5.1.1, 5.1.2 and 5.1.4 for details. * sections 5.1.1, 5.1.2 and 5.1.4 for details.
* *
* @param cls the Class to check, may be null * @param cls the Class to check, may be null
* @param toClass the Class to try to assign into, returns false if null * @param toClass the Class to try to assign into, returns false if null
* @param autoboxing whether to use implicit autoboxing/unboxing between primitives and wrappers * @param autoboxing whether to use implicit autoboxing/unboxing between primitives and wrappers
* @return {@code true} if assignment possible * @return {@code true} if assignment possible
*/ */
...@@ -914,7 +930,9 @@ public final class ExtractionUtils { ...@@ -914,7 +930,9 @@ public final class ExtractionUtils {
return toClass.isAssignableFrom(cls); return toClass.isAssignableFrom(cls);
} }
/** Maps primitive {@code Class}es to their corresponding wrapper {@code Class}. */ /**
* Maps primitive {@code Class}es to their corresponding wrapper {@code Class}.
*/
private static final Map<Class<?>, Class<?>> primitiveWrapperMap = new HashMap<>(); private static final Map<Class<?>, Class<?>> primitiveWrapperMap = new HashMap<>();
static { static {
...@@ -929,7 +947,9 @@ public final class ExtractionUtils { ...@@ -929,7 +947,9 @@ public final class ExtractionUtils {
primitiveWrapperMap.put(Void.TYPE, Void.TYPE); primitiveWrapperMap.put(Void.TYPE, Void.TYPE);
} }
/** Maps wrapper {@code Class}es to their corresponding primitive types. */ /**
* Maps wrapper {@code Class}es to their corresponding primitive types.
*/
private static final Map<Class<?>, Class<?>> wrapperPrimitiveMap = new HashMap<>(); private static final Map<Class<?>, Class<?>> wrapperPrimitiveMap = new HashMap<>();
static { static {
...@@ -948,7 +968,7 @@ public final class ExtractionUtils { ...@@ -948,7 +968,7 @@ public final class ExtractionUtils {
* *
* @param cls the class to convert, may be null * @param cls the class to convert, may be null
* @return the wrapper class for {@code cls} or {@code cls} if {@code cls} is not a primitive. * @return the wrapper class for {@code cls} or {@code cls} if {@code cls} is not a primitive.
* {@code null} if null input. * {@code null} if null input.
* @since 2.1 * @since 2.1
*/ */
public static Class<?> primitiveToWrapper(final Class<?> cls) { public static Class<?> primitiveToWrapper(final Class<?> cls) {
...@@ -969,7 +989,7 @@ public final class ExtractionUtils { ...@@ -969,7 +989,7 @@ public final class ExtractionUtils {
* *
* @param cls the class to convert, may be <b>null</b> * @param cls the class to convert, may be <b>null</b>
* @return the corresponding primitive type if {@code cls} is a wrapper class, <b>null</b> * @return the corresponding primitive type if {@code cls} is a wrapper class, <b>null</b>
* otherwise * otherwise
* @see #primitiveToWrapper(Class) * @see #primitiveToWrapper(Class)
* @since 2.4 * @since 2.4
*/ */
......
dlink-client/dlink-client-1.14/src/main/java/com/dlink/cdc/FlinkCDCMergeBuilder.java
View file @ f6d81b81
...@@ -21,10 +21,10 @@ import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer; ...@@ -21,10 +21,10 @@ import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
public class FlinkCDCMergeBuilder { public class FlinkCDCMergeBuilder {
public static void buildMySqlCDC(StreamExecutionEnvironment env, FlinkCDCConfig config) { public static void buildMySqlCDC(StreamExecutionEnvironment env, FlinkCDCConfig config) {
if(Asserts.isNotNull(config.getParallelism())){ if (Asserts.isNotNull(config.getParallelism())) {
env.setParallelism(config.getParallelism()); env.setParallelism(config.getParallelism());
} }
if(Asserts.isNotNull(config.getCheckpoint())){ if (Asserts.isNotNull(config.getCheckpoint())) {
env.enableCheckpointing(config.getCheckpoint()); env.enableCheckpointing(config.getCheckpoint());
} }
MySqlSourceBuilder<String> sourceBuilder = MySqlSource.<String>builder() MySqlSourceBuilder<String> sourceBuilder = MySqlSource.<String>builder()
...@@ -32,16 +32,16 @@ public class FlinkCDCMergeBuilder { ...@@ -32,16 +32,16 @@ public class FlinkCDCMergeBuilder {
.port(config.getPort()) .port(config.getPort())
.username(config.getUsername()) .username(config.getUsername())
.password(config.getPassword()); .password(config.getPassword());
if(Asserts.isNotNull(config.getDatabase())&&config.getDatabase().size()>0){ if (Asserts.isNotNull(config.getDatabase()) && config.getDatabase().size() > 0) {
sourceBuilder.databaseList(config.getDatabase().toArray(new String[0])); sourceBuilder.databaseList(config.getDatabase().toArray(new String[0]));
} }
if(Asserts.isNotNull(config.getTable())&&config.getTable().size()>0){ if (Asserts.isNotNull(config.getTable()) && config.getTable().size() > 0) {
sourceBuilder.tableList(config.getTable().toArray(new String[0])); sourceBuilder.tableList(config.getTable().toArray(new String[0]));
} }
MySqlSourceBuilder<String> builder = sourceBuilder MySqlSourceBuilder<String> builder = sourceBuilder
.deserializer(new JsonDebeziumDeserializationSchema()); .deserializer(new JsonDebeziumDeserializationSchema());
if(Asserts.isNotNullString(config.getStartupMode())){ if (Asserts.isNotNullString(config.getStartupMode())) {
switch (config.getStartupMode().toUpperCase()){ switch (config.getStartupMode().toUpperCase()) {
case "INITIAL": case "INITIAL":
builder.startupOptions(StartupOptions.initial()); builder.startupOptions(StartupOptions.initial());
break; break;
...@@ -54,12 +54,12 @@ public class FlinkCDCMergeBuilder { ...@@ -54,12 +54,12 @@ public class FlinkCDCMergeBuilder {
default: default:
builder.startupOptions(StartupOptions.latest()); builder.startupOptions(StartupOptions.latest());
} }
}else { } else {
builder.startupOptions(StartupOptions.latest()); builder.startupOptions(StartupOptions.latest());
} }
MySqlSource<String> sourceFunction = builder.build(); MySqlSource<String> sourceFunction = builder.build();
DataStreamSource<String> streamSource = env.fromSource(sourceFunction, WatermarkStrategy.noWatermarks(), "MySQL Source"); DataStreamSource<String> streamSource = env.fromSource(sourceFunction, WatermarkStrategy.noWatermarks(), "MySQL Source");
streamSource.addSink(getKafkaProducer(config.getBrokers(),config.getTopic())); streamSource.addSink(getKafkaProducer(config.getBrokers(), config.getTopic()));
} }
private static FlinkKafkaProducer<String> getKafkaProducer(String brokers, String topic) { private static FlinkKafkaProducer<String> getKafkaProducer(String brokers, String topic) {
......
dlink-client/dlink-client-1.14/src/main/java/com/dlink/executor/CustomTableEnvironmentImpl.java
View file @ f6d81b81
...@@ -298,7 +298,7 @@ public class CustomTableEnvironmentImpl extends TableEnvironmentImpl implements ...@@ -298,7 +298,7 @@ public class CustomTableEnvironmentImpl extends TableEnvironmentImpl implements
if (setOperation.getKey().isPresent() && setOperation.getValue().isPresent()) { if (setOperation.getKey().isPresent() && setOperation.getValue().isPresent()) {
String key = setOperation.getKey().get().trim(); String key = setOperation.getKey().get().trim();
String value = setOperation.getValue().get().trim(); String value = setOperation.getValue().get().trim();
if(Asserts.isNullString(key)||Asserts.isNullString(value)){ if (Asserts.isNullString(key) || Asserts.isNullString(value)) {
return; return;
} }
Map<String, String> confMap = new HashMap<>(); Map<String, String> confMap = new HashMap<>();
...@@ -313,7 +313,7 @@ public class CustomTableEnvironmentImpl extends TableEnvironmentImpl implements ...@@ -313,7 +313,7 @@ public class CustomTableEnvironmentImpl extends TableEnvironmentImpl implements
private void callReset(ResetOperation resetOperation, StreamExecutionEnvironment environment, Map<String, Object> setMap) { private void callReset(ResetOperation resetOperation, StreamExecutionEnvironment environment, Map<String, Object> setMap) {
if (resetOperation.getKey().isPresent()) { if (resetOperation.getKey().isPresent()) {
String key = resetOperation.getKey().get().trim(); String key = resetOperation.getKey().get().trim();
if(Asserts.isNullString(key)){ if (Asserts.isNullString(key)) {
return; return;
} }
Map<String, String> confMap = new HashMap<>(); Map<String, String> confMap = new HashMap<>();
......
dlink-client/dlink-client-1.14/src/main/java/com/dlink/executor/CustomTableResultImpl.java
View file @ f6d81b81
...@@ -17,22 +17,14 @@ import org.apache.flink.util.Preconditions; ...@@ -17,22 +17,14 @@ import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable; import javax.annotation.Nullable;
import java.io.PrintWriter; import java.io.PrintWriter;
import java.time.ZoneId; import java.time.ZoneId;
import java.util.ArrayList; import java.util.*;
import java.util.Collections; import java.util.concurrent.*;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/** /**
* 定制TableResultImpl * 定制TableResultImpl
* @author wenmo *
* @since 2021/10/22 10:02 * @author wenmo
* @since 2021/10/22 10:02
**/ **/
@Internal @Internal
public class CustomTableResultImpl implements TableResult { public class CustomTableResultImpl implements TableResult {
...@@ -68,16 +60,16 @@ public class CustomTableResultImpl implements TableResult { ...@@ -68,16 +60,16 @@ public class CustomTableResultImpl implements TableResult {
Preconditions.checkNotNull(sessionTimeZone, "sessionTimeZone should not be null"); Preconditions.checkNotNull(sessionTimeZone, "sessionTimeZone should not be null");
} }
public static TableResult buildTableResult(List<TableSchemaField> fields,List<Row> rows){ public static TableResult buildTableResult(List<TableSchemaField> fields, List<Row> rows) {
Builder builder = builder().resultKind(ResultKind.SUCCESS); Builder builder = builder().resultKind(ResultKind.SUCCESS);
if(fields.size()>0) { if (fields.size() > 0) {
List<String> columnNames = new ArrayList<>(); List<String> columnNames = new ArrayList<>();
List<DataType> columnTypes = new ArrayList<>(); List<DataType> columnTypes = new ArrayList<>();
for (int i = 0; i < fields.size(); i++) { for (int i = 0; i < fields.size(); i++) {
columnNames.add(fields.get(i).getName()); columnNames.add(fields.get(i).getName());
columnTypes.add(fields.get(i).getType()); columnTypes.add(fields.get(i).getType());
} }
builder.schema(ResolvedSchema.physical(columnNames,columnTypes)).data(rows); builder.schema(ResolvedSchema.physical(columnNames, columnTypes)).data(rows);
} }
return builder.build(); return builder.build();
} }
...@@ -184,7 +176,9 @@ public class CustomTableResultImpl implements TableResult { ...@@ -184,7 +176,9 @@ public class CustomTableResultImpl implements TableResult {
return new Builder(); return new Builder();
} }
/** Builder for creating a {@link CustomTableResultImpl}. */ /**
* Builder for creating a {@link CustomTableResultImpl}.
*/
public static class Builder { public static class Builder {
private JobClient jobClient = null; private JobClient jobClient = null;
private ResolvedSchema resolvedSchema = null; private ResolvedSchema resolvedSchema = null;
...@@ -194,7 +188,8 @@ public class CustomTableResultImpl implements TableResult { ...@@ -194,7 +188,8 @@ public class CustomTableResultImpl implements TableResult {
PrintStyle.tableau(Integer.MAX_VALUE, PrintUtils.NULL_COLUMN, false, false); PrintStyle.tableau(Integer.MAX_VALUE, PrintUtils.NULL_COLUMN, false, false);
private ZoneId sessionTimeZone = ZoneId.of("UTC"); private ZoneId sessionTimeZone = ZoneId.of("UTC");
private Builder() {} private Builder() {
}
/** /**
* Specifies job client which associates the submitted Flink job. * Specifies job client which associates the submitted Flink job.
...@@ -250,28 +245,36 @@ public class CustomTableResultImpl implements TableResult { ...@@ -250,28 +245,36 @@ public class CustomTableResultImpl implements TableResult {
return this; return this;
} }
/** Specifies print style. Default is {@link TableauStyle} with max integer column width. */ /**
* Specifies print style. Default is {@link TableauStyle} with max integer column width.
*/
public Builder setPrintStyle(PrintStyle printStyle) { public Builder setPrintStyle(PrintStyle printStyle) {
Preconditions.checkNotNull(printStyle, "printStyle should not be null"); Preconditions.checkNotNull(printStyle, "printStyle should not be null");
this.printStyle = printStyle; this.printStyle = printStyle;
return this; return this;
} }
/** Specifies session time zone. */ /**
* Specifies session time zone.
*/
public Builder setSessionTimeZone(ZoneId sessionTimeZone) { public Builder setSessionTimeZone(ZoneId sessionTimeZone) {
Preconditions.checkNotNull(sessionTimeZone, "sessionTimeZone should not be null"); Preconditions.checkNotNull(sessionTimeZone, "sessionTimeZone should not be null");
this.sessionTimeZone = sessionTimeZone; this.sessionTimeZone = sessionTimeZone;
return this; return this;
} }
/** Returns a {@link TableResult} instance. */ /**
* Returns a {@link TableResult} instance.
*/
public TableResult build() { public TableResult build() {
return new CustomTableResultImpl( return new CustomTableResultImpl(
jobClient, resolvedSchema, resultKind, data, printStyle, sessionTimeZone); jobClient, resolvedSchema, resultKind, data, printStyle, sessionTimeZone);
} }
} }
/** Root interface for all print styles. */ /**
* Root interface for all print styles.
*/
public interface PrintStyle { public interface PrintStyle {
/** /**
* Create a tableau print style with given max column width, null column, change mode * Create a tableau print style with given max column width, null column, change mode
...@@ -299,7 +302,9 @@ public class CustomTableResultImpl implements TableResult { ...@@ -299,7 +302,9 @@ public class CustomTableResultImpl implements TableResult {
} }
} }
/** print the result schema and content as tableau form. */ /**
* print the result schema and content as tableau form.
*/
private static final class TableauStyle implements PrintStyle { private static final class TableauStyle implements PrintStyle {
/** /**
* A flag to indicate whether the column width is derived from type (true) or content * A flag to indicate whether the column width is derived from type (true) or content
...@@ -309,7 +314,9 @@ public class CustomTableResultImpl implements TableResult { ...@@ -309,7 +314,9 @@ public class CustomTableResultImpl implements TableResult {
private final int maxColumnWidth; private final int maxColumnWidth;
private final String nullColumn; private final String nullColumn;
/** A flag to indicate whether print row kind info. */ /**
* A flag to indicate whether print row kind info.
*/
private final boolean printRowKind; private final boolean printRowKind;
private TableauStyle( private TableauStyle(
...@@ -343,7 +350,8 @@ public class CustomTableResultImpl implements TableResult { ...@@ -343,7 +350,8 @@ public class CustomTableResultImpl implements TableResult {
/** /**
* only print the result content as raw form. column delimiter is ",", row delimiter is "\n". * only print the result content as raw form. column delimiter is ",", row delimiter is "\n".
*/ */
private static final class RawContentStyle implements PrintStyle {} private static final class RawContentStyle implements PrintStyle {
}
/** /**
* A {@link CloseableIterator} wrapper class that can return whether the first row is ready. * A {@link CloseableIterator} wrapper class that can return whether the first row is ready.
......
dlink-client/dlink-client-1.14/src/main/java/com/dlink/executor/TableSchemaField.java
View file @ f6d81b81
...@@ -3,8 +3,8 @@ package com.dlink.executor; ...@@ -3,8 +3,8 @@ package com.dlink.executor;
import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.DataType;
/** /**
* @author wenmo * @author wenmo
* @since 2021/10/22 10:02 * @since 2021/10/22 10:02
**/ **/
public class TableSchemaField { public class TableSchemaField {
private String name; private String name;
......
dlink-client/dlink-client-1.14/src/main/java/com/dlink/utils/FlinkUtil.java
View file @ f6d81b81
...@@ -16,19 +16,19 @@ import java.util.Optional; ...@@ -16,19 +16,19 @@ import java.util.Optional;
*/ */
public class FlinkUtil { public class FlinkUtil {
public static List<String> getFieldNamesFromCatalogManager(CatalogManager catalogManager, String catalog, String database, String table){ public static List<String> getFieldNamesFromCatalogManager(CatalogManager catalogManager, String catalog, String database, String table) {
Optional<CatalogManager.TableLookupResult> tableOpt = catalogManager.getTable( Optional<CatalogManager.TableLookupResult> tableOpt = catalogManager.getTable(
ObjectIdentifier.of(catalog, database, table) ObjectIdentifier.of(catalog, database, table)
); );
if (tableOpt.isPresent()) { if (tableOpt.isPresent()) {
return tableOpt.get().getResolvedSchema().getColumnNames(); return tableOpt.get().getResolvedSchema().getColumnNames();
}else{ } else {
return new ArrayList<String>(); return new ArrayList<String>();
} }
} }
public static List<String> catchColumn(TableResult tableResult){ public static List<String> catchColumn(TableResult tableResult) {
return tableResult.getResolvedSchema().getColumnNames(); return tableResult.getResolvedSchema().getColumnNames();
} }
} }
dlink-client/dlink-client-1.14/src/main/java/org/apache/flink/table/types/extraction/ExtractionUtils.java
View file @ f6d81b81
...@@ -21,42 +21,20 @@ package org.apache.flink.table.types.extraction; ...@@ -21,42 +21,20 @@ package org.apache.flink.table.types.extraction;
import com.dlink.pool.ClassPool; import com.dlink.pool.ClassPool;
import org.apache.flink.annotation.Internal; import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.shaded.asm7.org.objectweb.asm.*;
import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.DataTypeFactory; import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.StructuredType; import org.apache.flink.table.types.logical.StructuredType;
import org.apache.flink.shaded.asm7.org.objectweb.asm.ClassReader;
import org.apache.flink.shaded.asm7.org.objectweb.asm.ClassVisitor;
import org.apache.flink.shaded.asm7.org.objectweb.asm.Label;
import org.apache.flink.shaded.asm7.org.objectweb.asm.MethodVisitor;
import org.apache.flink.shaded.asm7.org.objectweb.asm.Opcodes;
import javax.annotation.Nullable; import javax.annotation.Nullable;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
import java.lang.annotation.Annotation; import java.lang.annotation.Annotation;
import java.lang.reflect.Constructor;
import java.lang.reflect.Executable;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.lang.reflect.Parameter;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type; import java.lang.reflect.Type;
import java.lang.reflect.TypeVariable; import java.lang.reflect.*;
import java.util.ArrayList; import java.util.*;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.function.Function; import java.util.function.Function;
import java.util.regex.Pattern; import java.util.regex.Pattern;
import java.util.stream.Collectors; import java.util.stream.Collectors;
...@@ -65,7 +43,9 @@ import java.util.stream.Stream; ...@@ -65,7 +43,9 @@ import java.util.stream.Stream;
import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getConstructorDescriptor; import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getConstructorDescriptor;
import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getMethodDescriptor; import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getMethodDescriptor;
/** Utilities for performing reflection tasks. */ /**
* Utilities for performing reflection tasks.
*/
@Internal @Internal
public final class ExtractionUtils { public final class ExtractionUtils {
...@@ -73,7 +53,9 @@ public final class ExtractionUtils { ...@@ -73,7 +53,9 @@ public final class ExtractionUtils {
// Methods shared across packages // Methods shared across packages
// -------------------------------------------------------------------------------------------- // --------------------------------------------------------------------------------------------
/** Collects methods of the given name. */ /**
* Collects methods of the given name.
*/
public static List<Method> collectMethods(Class<?> function, String methodName) { public static List<Method> collectMethods(Class<?> function, String methodName) {
return Arrays.stream(function.getMethods()) return Arrays.stream(function.getMethods())
.filter(method -> method.getName().equals(methodName)) .filter(method -> method.getName().equals(methodName))
...@@ -133,7 +115,9 @@ public final class ExtractionUtils { ...@@ -133,7 +115,9 @@ public final class ExtractionUtils {
return clz == null || ExtractionUtils.isAssignable(clz, param, true); return clz == null || ExtractionUtils.isAssignable(clz, param, true);
} }
/** Creates a method signature string like {@code int eval(Integer, String)}. */ /**
* Creates a method signature string like {@code int eval(Integer, String)}.
*/
public static String createMethodSignatureString( public static String createMethodSignatureString(
String methodName, Class<?>[] parameters, @Nullable Class<?> returnType) { String methodName, Class<?>[] parameters, @Nullable Class<?> returnType) {
final StringBuilder builder = new StringBuilder(); final StringBuilder builder = new StringBuilder();
...@@ -306,7 +290,9 @@ public final class ExtractionUtils { ...@@ -306,7 +290,9 @@ public final class ExtractionUtils {
return false; return false;
} }
/** Checks whether a field is directly readable without a getter. */ /**
* Checks whether a field is directly readable without a getter.
*/
public static boolean isStructuredFieldDirectlyReadable(Field field) { public static boolean isStructuredFieldDirectlyReadable(Field field) {
final int m = field.getModifiers(); final int m = field.getModifiers();
...@@ -314,7 +300,9 @@ public final class ExtractionUtils { ...@@ -314,7 +300,9 @@ public final class ExtractionUtils {
return Modifier.isPublic(m); return Modifier.isPublic(m);
} }
/** Checks whether a field is directly writable without a setter or constructor. */ /**
* Checks whether a field is directly writable without a setter or constructor.
*/
public static boolean isStructuredFieldDirectlyWritable(Field field) { public static boolean isStructuredFieldDirectlyWritable(Field field) {
final int m = field.getModifiers(); final int m = field.getModifiers();
...@@ -353,12 +341,16 @@ public final class ExtractionUtils { ...@@ -353,12 +341,16 @@ public final class ExtractionUtils {
// Methods intended for this package // Methods intended for this package
// -------------------------------------------------------------------------------------------- // --------------------------------------------------------------------------------------------
/** Helper method for creating consistent exceptions during extraction. */ /**
* Helper method for creating consistent exceptions during extraction.
*/
static ValidationException extractionError(String message, Object... args) { static ValidationException extractionError(String message, Object... args) {
return extractionError(null, message, args); return extractionError(null, message, args);
} }
/** Helper method for creating consistent exceptions during extraction. */ /**
* Helper method for creating consistent exceptions during extraction.
*/
static ValidationException extractionError(Throwable cause, String message, Object... args) { static ValidationException extractionError(Throwable cause, String message, Object... args) {
return new ValidationException(String.format(message, args), cause); return new ValidationException(String.format(message, args), cause);
} }
...@@ -387,8 +379,11 @@ public final class ExtractionUtils { ...@@ -387,8 +379,11 @@ public final class ExtractionUtils {
return typeHierarchy; return typeHierarchy;
} }
/** Converts a {@link Type} to {@link Class} if possible, {@code null} otherwise. */ /**
static @Nullable Class<?> toClass(Type type) { * Converts a {@link Type} to {@link Class} if possible, {@code null} otherwise.
*/
static @Nullable
Class<?> toClass(Type type) {
if (type instanceof Class) { if (type instanceof Class) {
return (Class<?>) type; return (Class<?>) type;
} else if (type instanceof ParameterizedType) { } else if (type instanceof ParameterizedType) {
...@@ -399,7 +394,9 @@ public final class ExtractionUtils { ...@@ -399,7 +394,9 @@ public final class ExtractionUtils {
return null; return null;
} }
/** Creates a raw data type. */ /**
* Creates a raw data type.
*/
@SuppressWarnings({"unchecked", "rawtypes"}) @SuppressWarnings({"unchecked", "rawtypes"})
static DataType createRawType( static DataType createRawType(
DataTypeFactory typeFactory, DataTypeFactory typeFactory,
...@@ -433,7 +430,9 @@ public final class ExtractionUtils { ...@@ -433,7 +430,9 @@ public final class ExtractionUtils {
} }
} }
/** Resolves a {@link TypeVariable} using the given type hierarchy if possible. */ /**
* Resolves a {@link TypeVariable} using the given type hierarchy if possible.
*/
static Type resolveVariable(List<Type> typeHierarchy, TypeVariable<?> variable) { static Type resolveVariable(List<Type> typeHierarchy, TypeVariable<?> variable) {
// iterate through hierarchy from top to bottom until type variable gets a non-variable // iterate through hierarchy from top to bottom until type variable gets a non-variable
// assigned // assigned
...@@ -456,7 +455,8 @@ public final class ExtractionUtils { ...@@ -456,7 +455,8 @@ public final class ExtractionUtils {
return variable; return variable;
} }
private static @Nullable Type resolveVariableInParameterizedType( private static @Nullable
Type resolveVariableInParameterizedType(
TypeVariable<?> variable, ParameterizedType currentType) { TypeVariable<?> variable, ParameterizedType currentType) {
final Class<?> currentRaw = (Class<?>) currentType.getRawType(); final Class<?> currentRaw = (Class<?>) currentType.getRawType();
final TypeVariable<?>[] currentVariables = currentRaw.getTypeParameters(); final TypeVariable<?>[] currentVariables = currentRaw.getTypeParameters();
...@@ -494,7 +494,9 @@ public final class ExtractionUtils { ...@@ -494,7 +494,9 @@ public final class ExtractionUtils {
} }
} }
/** Returns the fields of a class for a {@link StructuredType}. */ /**
* Returns the fields of a class for a {@link StructuredType}.
*/
static List<Field> collectStructuredFields(Class<?> clazz) { static List<Field> collectStructuredFields(Class<?> clazz) {
final List<Field> fields = new ArrayList<>(); final List<Field> fields = new ArrayList<>();
while (clazz != Object.class) { while (clazz != Object.class) {
...@@ -511,7 +513,9 @@ public final class ExtractionUtils { ...@@ -511,7 +513,9 @@ public final class ExtractionUtils {
return fields; return fields;
} }
/** Validates if a field is properly readable either directly or through a getter. */ /**
* Validates if a field is properly readable either directly or through a getter.
*/
static void validateStructuredFieldReadability(Class<?> clazz, Field field) { static void validateStructuredFieldReadability(Class<?> clazz, Field field) {
// field is accessible // field is accessible
if (isStructuredFieldDirectlyReadable(field)) { if (isStructuredFieldDirectlyReadable(field)) {
...@@ -554,7 +558,9 @@ public final class ExtractionUtils { ...@@ -554,7 +558,9 @@ public final class ExtractionUtils {
field.getName(), clazz.getName()); field.getName(), clazz.getName());
} }
/** Returns the boxed type of a primitive type. */ /**
* Returns the boxed type of a primitive type.
*/
static Type primitiveToWrapper(Type type) { static Type primitiveToWrapper(Type type) {
if (type instanceof Class) { if (type instanceof Class) {
return primitiveToWrapper((Class<?>) type); return primitiveToWrapper((Class<?>) type);
...@@ -562,7 +568,9 @@ public final class ExtractionUtils { ...@@ -562,7 +568,9 @@ public final class ExtractionUtils {
return type; return type;
} }
/** Collects all methods that qualify as methods of a {@link StructuredType}. */ /**
* Collects all methods that qualify as methods of a {@link StructuredType}.
*/
static List<Method> collectStructuredMethods(Class<?> clazz) { static List<Method> collectStructuredMethods(Class<?> clazz) {
final List<Method> methods = new ArrayList<>(); final List<Method> methods = new ArrayList<>();
while (clazz != Object.class) { while (clazz != Object.class) {
...@@ -613,7 +621,9 @@ public final class ExtractionUtils { ...@@ -613,7 +621,9 @@ public final class ExtractionUtils {
// Parameter Extraction Utilities // Parameter Extraction Utilities
// -------------------------------------------------------------------------------------------- // --------------------------------------------------------------------------------------------
/** Result of the extraction in {@link #extractAssigningConstructor(Class, List)}. */ /**
* Result of the extraction in {@link #extractAssigningConstructor(Class, List)}.
*/
public static class AssigningConstructor { public static class AssigningConstructor {
public final Constructor<?> constructor; public final Constructor<?> constructor;
public final List<String> parameterNames; public final List<String> parameterNames;
...@@ -628,7 +638,8 @@ public final class ExtractionUtils { ...@@ -628,7 +638,8 @@ public final class ExtractionUtils {
* Checks whether the given constructor takes all of the given fields with matching (possibly * Checks whether the given constructor takes all of the given fields with matching (possibly
* primitive) type and name. An assigning constructor can define the order of fields. * primitive) type and name. An assigning constructor can define the order of fields.
*/ */
public static @Nullable AssigningConstructor extractAssigningConstructor( public static @Nullable
AssigningConstructor extractAssigningConstructor(
Class<?> clazz, List<Field> fields) { Class<?> clazz, List<Field> fields) {
AssigningConstructor foundConstructor = null; AssigningConstructor foundConstructor = null;
for (Constructor<?> constructor : clazz.getDeclaredConstructors()) { for (Constructor<?> constructor : clazz.getDeclaredConstructors()) {
...@@ -652,8 +663,11 @@ public final class ExtractionUtils { ...@@ -652,8 +663,11 @@ public final class ExtractionUtils {
return foundConstructor; return foundConstructor;
} }
/** Extracts the parameter names of a method if possible. */ /**
static @Nullable List<String> extractMethodParameterNames(Method method) { * Extracts the parameter names of a method if possible.
*/
static @Nullable
List<String> extractMethodParameterNames(Method method) {
return extractExecutableNames(method); return extractExecutableNames(method);
} }
...@@ -661,7 +675,8 @@ public final class ExtractionUtils { ...@@ -661,7 +675,8 @@ public final class ExtractionUtils {
* Extracts ordered parameter names from a constructor that takes all of the given fields with * Extracts ordered parameter names from a constructor that takes all of the given fields with
* matching (possibly primitive and lenient) type and name. * matching (possibly primitive and lenient) type and name.
*/ */
private static @Nullable List<String> extractConstructorParameterNames( private static @Nullable
List<String> extractConstructorParameterNames(
Constructor<?> constructor, List<Field> fields) { Constructor<?> constructor, List<Field> fields) {
final Type[] parameterTypes = constructor.getGenericParameterTypes(); final Type[] parameterTypes = constructor.getGenericParameterTypes();
...@@ -698,7 +713,8 @@ public final class ExtractionUtils { ...@@ -698,7 +713,8 @@ public final class ExtractionUtils {
return fieldNames; return fieldNames;
} }
private static @Nullable List<String> extractExecutableNames(Executable executable) { private static @Nullable
List<String> extractExecutableNames(Executable executable) {
final int offset; final int offset;
if (!Modifier.isStatic(executable.getModifiers())) { if (!Modifier.isStatic(executable.getModifiers())) {
// remove "this" as first parameter // remove "this" as first parameter
...@@ -744,7 +760,7 @@ public final class ExtractionUtils { ...@@ -744,7 +760,7 @@ public final class ExtractionUtils {
private static ClassReader getClassReader(Class<?> cls) { private static ClassReader getClassReader(Class<?> cls) {
final String className = cls.getName().replaceFirst("^.*\\.", "") + ".class"; final String className = cls.getName().replaceFirst("^.*\\.", "") + ".class";
if(ClassPool.exist(cls.getName())){ if (ClassPool.exist(cls.getName())) {
return new ClassReader(ClassPool.get(cls.getName()).getClassByte()); return new ClassReader(ClassPool.get(cls.getName()).getClassByte());
} }
try (InputStream i = cls.getResourceAsStream(className)) { try (InputStream i = cls.getResourceAsStream(className)) {
...@@ -837,8 +853,8 @@ public final class ExtractionUtils { ...@@ -837,8 +853,8 @@ public final class ExtractionUtils {
* href="http://docs.oracle.com/javase/specs/">The Java Language Specification</a></em>, * href="http://docs.oracle.com/javase/specs/">The Java Language Specification</a></em>,
* sections 5.1.1, 5.1.2 and 5.1.4 for details. * sections 5.1.1, 5.1.2 and 5.1.4 for details.
* *
* @param cls the Class to check, may be null * @param cls the Class to check, may be null
* @param toClass the Class to try to assign into, returns false if null * @param toClass the Class to try to assign into, returns false if null
* @param autoboxing whether to use implicit autoboxing/unboxing between primitives and wrappers * @param autoboxing whether to use implicit autoboxing/unboxing between primitives and wrappers
* @return {@code true} if assignment possible * @return {@code true} if assignment possible
*/ */
...@@ -915,7 +931,9 @@ public final class ExtractionUtils { ...@@ -915,7 +931,9 @@ public final class ExtractionUtils {
return toClass.isAssignableFrom(cls); return toClass.isAssignableFrom(cls);
} }
/** Maps primitive {@code Class}es to their corresponding wrapper {@code Class}. */ /**
* Maps primitive {@code Class}es to their corresponding wrapper {@code Class}.
*/
private static final Map<Class<?>, Class<?>> primitiveWrapperMap = new HashMap<>(); private static final Map<Class<?>, Class<?>> primitiveWrapperMap = new HashMap<>();
static { static {
...@@ -930,7 +948,9 @@ public final class ExtractionUtils { ...@@ -930,7 +948,9 @@ public final class ExtractionUtils {
primitiveWrapperMap.put(Void.TYPE, Void.TYPE); primitiveWrapperMap.put(Void.TYPE, Void.TYPE);
} }
/** Maps wrapper {@code Class}es to their corresponding primitive types. */ /**
* Maps wrapper {@code Class}es to their corresponding primitive types.
*/
private static final Map<Class<?>, Class<?>> wrapperPrimitiveMap = new HashMap<>(); private static final Map<Class<?>, Class<?>> wrapperPrimitiveMap = new HashMap<>();
static { static {
...@@ -949,7 +969,7 @@ public final class ExtractionUtils { ...@@ -949,7 +969,7 @@ public final class ExtractionUtils {
* *
* @param cls the class to convert, may be null * @param cls the class to convert, may be null
* @return the wrapper class for {@code cls} or {@code cls} if {@code cls} is not a primitive. * @return the wrapper class for {@code cls} or {@code cls} if {@code cls} is not a primitive.
* {@code null} if null input. * {@code null} if null input.
* @since 2.1 * @since 2.1
*/ */
public static Class<?> primitiveToWrapper(final Class<?> cls) { public static Class<?> primitiveToWrapper(final Class<?> cls) {
...@@ -970,7 +990,7 @@ public final class ExtractionUtils { ...@@ -970,7 +990,7 @@ public final class ExtractionUtils {
* *
* @param cls the class to convert, may be <b>null</b> * @param cls the class to convert, may be <b>null</b>
* @return the corresponding primitive type if {@code cls} is a wrapper class, <b>null</b> * @return the corresponding primitive type if {@code cls} is a wrapper class, <b>null</b>
* otherwise * otherwise
* @see #primitiveToWrapper(Class) * @see #primitiveToWrapper(Class)
* @since 2.4 * @since 2.4
*/ */
......
dlink-client/dlink-client-base/src/main/java/com/dlink/utils/FlinkBaseUtil.java
View file @ f6d81b81
...@@ -14,14 +14,14 @@ import java.util.Map; ...@@ -14,14 +14,14 @@ import java.util.Map;
*/ */
public class FlinkBaseUtil { public class FlinkBaseUtil {
public static Map<String,String> getParamsFromArgs(String[] args){ public static Map<String, String> getParamsFromArgs(String[] args) {
Map<String,String> params = new HashMap<>(); Map<String, String> params = new HashMap<>();
ParameterTool parameters = ParameterTool.fromArgs(args); ParameterTool parameters = ParameterTool.fromArgs(args);
params.put(FlinkParamConstant.ID,parameters.get(FlinkParamConstant.ID, null)); params.put(FlinkParamConstant.ID, parameters.get(FlinkParamConstant.ID, null));
params.put(FlinkParamConstant.DRIVER,parameters.get(FlinkParamConstant.DRIVER, null)); params.put(FlinkParamConstant.DRIVER, parameters.get(FlinkParamConstant.DRIVER, null));
params.put(FlinkParamConstant.URL,parameters.get(FlinkParamConstant.URL, null)); params.put(FlinkParamConstant.URL, parameters.get(FlinkParamConstant.URL, null));
params.put(FlinkParamConstant.USERNAME,parameters.get(FlinkParamConstant.USERNAME, null)); params.put(FlinkParamConstant.USERNAME, parameters.get(FlinkParamConstant.USERNAME, null));
params.put(FlinkParamConstant.PASSWORD,parameters.get(FlinkParamConstant.PASSWORD, null)); params.put(FlinkParamConstant.PASSWORD, parameters.get(FlinkParamConstant.PASSWORD, null));
return params; return params;
} }
} }
dlink-client/dlink-client-base/src/main/resources/spring/dubbo-demo-consumer.xml deleted 100644 → 0
View file @ 2a471036
<?xml version="1.0" encoding="UTF-8"?>
<!--
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF licenses this file to You under the Apache License, Version 2.0
(the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
-->
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:dubbo="http://dubbo.apache.org/schema/dubbo"
xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<context:property-placeholder/>
<dubbo:application name="demo-consumer"/>
<dubbo:registry address="zookeeper://${zookeeper.address:127.0.0.1}:2181"/>
<!-- <dubbo:reference id="demoService" check="true" interface="com.dlink.service.DemoService" version="1.0.0"/>-->
</beans>
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment