From c15757ba38755089a8929ba33964578ff6fcfd5e Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Tue, 12 Apr 2022 20:31:27 +0800 Subject: [PATCH 01/17] SinkTask and SourceTask implement the validate method https://github.com/apache/rocketmq-connect/issues/85 --- .../connector/api/component/task/sink/SinkTask.java | 10 ++++++++++ .../api/component/task/source/SourceTask.java | 10 ++++++++++ 2 files changed, 20 insertions(+) diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java b/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java index 0268397..fda1207 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java @@ -14,6 +14,7 @@ package io.openmessaging.connector.api.component.task.sink; +import io.openmessaging.KeyValue; import io.openmessaging.connector.api.component.task.Task; import io.openmessaging.connector.api.data.ConnectRecord; import io.openmessaging.connector.api.data.RecordOffset; @@ -57,4 +58,13 @@ public Map preCommit(Map Date: Fri, 15 Apr 2022 17:00:53 +0800 Subject: [PATCH 02/17] Adjust the init and start methods of the component interface --- .../connector/api/component/Component.java | 8 ++++---- .../api/component/connector/Connector.java | 2 +- .../api/component/task/sink/SinkTask.java | 2 +- .../api/component/task/source/SourceTask.java | 17 ++++------------- 4 files changed, 10 insertions(+), 19 deletions(-) diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/Component.java b/connector/src/main/java/io/openmessaging/connector/api/component/Component.java index 0fed332..0349c2f 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/Component.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/Component.java @@ -28,16 +28,16 @@ public interface Component { /** * Init the component * - * @param config component config + * @param context component context */ - void init(KeyValue config); + void init(R context); /** * Start the component * - * @param componentContext component context + * @param config component context */ - void start(R componentContext); + void start(KeyValue config); /** * Stop the component. diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/connector/Connector.java b/connector/src/main/java/io/openmessaging/connector/api/component/connector/Connector.java index dd8ac3e..d9272a5 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/connector/Connector.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/connector/Connector.java @@ -23,7 +23,7 @@ public abstract class Connector implements Component { protected ConnectorContext connectorContext; - @Override public void start(ConnectorContext connectorContext) { + @Override public void init(ConnectorContext connectorContext) { this.connectorContext = connectorContext; } diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java b/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java index fda1207..c26d482 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java @@ -27,7 +27,7 @@ public abstract class SinkTask implements Task { protected SinkTaskContext sinkTaskContext; - @Override public void start(SinkTaskContext sinkTaskContext) { + @Override public void init(SinkTaskContext sinkTaskContext) { this.sinkTaskContext = sinkTaskContext; } diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java b/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java index 8e07abc..5fec1a9 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java @@ -22,8 +22,7 @@ public abstract class SourceTask implements Task { protected SourceTaskContext sourceTaskContext; - - @Override public void start(SourceTaskContext sourceTaskContext) { + @Override public void init(SourceTaskContext sourceTaskContext) { this.sourceTaskContext = sourceTaskContext; } @@ -46,7 +45,6 @@ public abstract class SourceTask implements Task { *

* * @throws InterruptedException task thread interupt exception - * * @param connectRecords connect records */ public void commit(final List connectRecords) throws InterruptedException { @@ -57,25 +55,18 @@ public void commit(final List connectRecords) throws InterruptedE * If the user wants to use external storage to save the position,user can implement this * function. */ - public void commit() { - } + public void commit() { } /** * Get source task context. - * * @return source task context */ + @Deprecated public SourceTaskContext getContext() { return sourceTaskContext; } - /** - * Should invoke before start the connector. - * @param config component config - */ @Override - public void validate(KeyValue config) { - - } + public void validate(KeyValue config){} } From fa134ee3954b3e130d0052ecfd48aa89ec512180 Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Fri, 15 Apr 2022 17:17:37 +0800 Subject: [PATCH 03/17] Set pause and resume to deprecated methods. It feels like they can be removed --- .../connector/api/component/task/Task.java | 2 ++ .../api/component/task/sink/SinkTask.java | 16 +++++++++++++++ .../api/component/task/source/SourceTask.java | 20 ++++++++++++++++++- 3 files changed, 37 insertions(+), 1 deletion(-) diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/task/Task.java b/connector/src/main/java/io/openmessaging/connector/api/component/task/Task.java index 3254e1c..5c34f59 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/task/Task.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/task/Task.java @@ -22,11 +22,13 @@ public interface Task extends Component { /** * Pause the task. */ + @Deprecated void pause(); /** * Resume the task. */ + @Deprecated void resume(); } diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java b/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java index c26d482..9fc8f7a 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java @@ -67,4 +67,20 @@ public Map preCommit(Map Date: Tue, 24 May 2022 21:44:05 +0800 Subject: [PATCH 04/17] Add struct object and optimize schema and schema builder API #41 --- .../connector/api/data/FieldType.java | 115 +++++++- .../connector/api/data/Schema.java | 234 ++++++++++++++-- .../connector/api/data/SchemaBuilder.java | 193 ++++++++++++- .../connector/api/data/Struct.java | 261 ++++++++++++++++++ .../api/errors/SchemaBuilderException.java | 28 ++ 5 files changed, 789 insertions(+), 42 deletions(-) create mode 100644 connector/src/main/java/io/openmessaging/connector/api/data/Struct.java create mode 100644 connector/src/main/java/io/openmessaging/connector/api/errors/SchemaBuilderException.java diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/FieldType.java b/connector/src/main/java/io/openmessaging/connector/api/data/FieldType.java index 9eb1009..0ff6467 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/data/FieldType.java +++ b/connector/src/main/java/io/openmessaging/connector/api/data/FieldType.java @@ -14,47 +14,132 @@ package io.openmessaging.connector.api.data; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + /** * Define the field type. */ public enum FieldType { - /** Short Integer */ + + /** + * Byte Integer + */ INT8, - /** Integer */ + /** + * Short + */ + INT16, + + /** + * Integer + */ INT32, - /** Long */ + /** + * Long + */ INT64, - /** BigInteger */ - BIG_INTEGER, - - /** Float */ + /** + * Float + */ FLOAT32, - /** Double */ + /** + * Double + */ FLOAT64, - /** Boolean */ + /** + * Boolean + */ BOOLEAN, - /** String */ + /** + * String + */ STRING, - /** Byte */ + /** + * Byte + */ BYTES, - /** List */ + /** + * List + */ ARRAY, - /** Map */ + /** + * Map + */ MAP, - /** Date */ + /** + * Date + */ DATETIME, + /** * Structured */ - STRUCT + STRUCT; + + /** + * Maps Schema.Types to a list of Java classes that can be used to represent them. + */ + public static final Map> SCHEMA_TYPE_CLASSES = new EnumMap<>(FieldType.class); + /** + * Maps the Java classes to the corresponding Schema.Type. + */ + public static final Map, FieldType> JAVA_CLASS_SCHEMA_TYPES = new HashMap<>(); + + static { + SCHEMA_TYPE_CLASSES.put(FieldType.INT8, Collections.singletonList((Class) Byte.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.INT16, Collections.singletonList((Class) Short.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.INT32, Collections.singletonList((Class) Integer.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.INT64, Collections.singletonList((Class) Long.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.FLOAT32, Collections.singletonList((Class) Float.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.FLOAT64, Collections.singletonList((Class) Double.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.BOOLEAN, Collections.singletonList((Class) Boolean.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.STRING, Collections.singletonList((Class) String.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.BYTES, Arrays.asList((Class) byte[].class, (Class) ByteBuffer.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.ARRAY, Collections.singletonList((Class) List.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.MAP, Collections.singletonList((Class) Map.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.STRUCT, Collections.singletonList((Class) Struct.class)); + + for (Map.Entry> schemaClasses : SCHEMA_TYPE_CLASSES.entrySet()) { + for (Class schemaClass : schemaClasses.getValue()) { + JAVA_CLASS_SCHEMA_TYPES.put(schemaClass, schemaClasses.getKey()); + } + } + } + + /** + * Determine whether it is a basic type + * + * @return + */ + public boolean isPrimitive() { + switch (this) { + case INT8: + case INT16: + case INT32: + case INT64: + case FLOAT32: + case FLOAT64: + case BOOLEAN: + case STRING: + case BYTES: + return true; + } + return false; + } } diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java b/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java index 5db07c6..f4174ea 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java +++ b/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java @@ -14,7 +14,12 @@ package io.openmessaging.connector.api.data; +import io.openmessaging.connector.api.errors.ConnectException; + +import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; /** * Schema @@ -25,7 +30,22 @@ public class Schema { * Name of the schema. */ private String name; - + /** + * schema version + */ + private Integer version; + /** + * optional + */ + private boolean optional; + /** + * default value + */ + private Object defaultValue; + /** + * Description schema + */ + private String doc; /** * Type of the field */ @@ -34,13 +54,106 @@ public class Schema { * Structure of the schema, contains a list of {@link Field} */ private List fields; + private Map fieldsByName; - public Schema(String name, FieldType fieldType, List fields) { - this.name = name; + /** + * map type + */ + private Schema keySchema; + private Schema valueSchema; + + /** + * Construct a Schema. Most users should not construct schemas manually, preferring {@link SchemaBuilder} instead. + */ + public Schema(String name,FieldType fieldType, boolean optional, Object defaultValue, Integer version, String doc, List fields,Schema keySchema, Schema valueSchema) { this.fieldType = fieldType; - this.fields = fields; + this.optional = optional; + this.defaultValue = defaultValue; + this.name = name; + this.version = version; + this.doc = doc; + if (this.fieldType == FieldType.STRUCT) { + this.fields = fields == null ? Collections.emptyList() : fields; + this.fieldsByName = new HashMap<>(this.fields.size()); + for (Field field : this.fields) { + fieldsByName.put(field.getName(), field); + } + } else { + this.fields = null; + this.fieldsByName = null; + } + this.keySchema = keySchema; + this.valueSchema = valueSchema; + } + + + /** + * Construct a Schema. Most users should not construct schemas manually, preferring {@link SchemaBuilder} instead. + */ + public Schema(String name,FieldType fieldType, boolean optional, Object defaultValue, Integer version, List fields) { + this(name, fieldType, optional, defaultValue, version, null, fields, null, null); + } + + + /** + * Construct a Schema. Most users should not construct schemas manually, preferring {@link SchemaBuilder} instead. + */ + public Schema(String name, FieldType fieldType, List fields) { + this(name, fieldType, false, null, null, fields); + } + + /** + * get field + * @param fieldName + * @return + */ + public Field getField(String fieldName) { + if (fieldsByName.containsKey(fieldName)){ + return fieldsByName.get(fieldName); + } + return null; + } + + /** + * add field + * @param field + */ + public void addField(Field field) { + this.fields.add(field); + this.fieldsByName.put(field.getName(), field); + } + + public Schema keySchema() { + return keySchema; + } + public Schema valueSchema() { + return valueSchema; } + public boolean isOptional() { + return optional; + } + + public Object defaultValue(){ + return defaultValue; + } + + /** + * Get the optional version of the schema. If a version is included, newer versions *must* be larger than older ones. + * @return the version of this schema + */ + public Integer version(){ + return version; + } + + /** + * @return the documentation for this schema + */ + public String doc() { + return doc; + } + + public String getName() { return name; } @@ -49,6 +162,14 @@ public void setName(String name) { this.name = name; } + public FieldType getFieldType() { + return fieldType; + } + + public void setFieldType(FieldType fieldType) { + this.fieldType = fieldType; + } + public List getFields() { return fields; } @@ -57,34 +178,101 @@ public void setFields(List fields) { this.fields = fields; } - public Field getField(String fieldName) { + /** + * Validate that the value can be used for this schema + */ + public void validateValue(Object value) { + validateValue(this, value); + } + /** + * Validate that the value can be used with the schema + */ + public static void validateValue(Schema schema, Object value) { + validateValue(null, schema, value); + } + public static void validateValue(String name, Schema schema, Object value) { - for (Field field : fields) { - if (field.getName() - .equals(fieldName)) { - return field; + // check optional + if (value == null) { + if (!schema.isOptional()) { + throw new ConnectException("Invalid value: null used for required field: \"" + name + + "\", schema type: " + schema.getFieldType()); } + return; } - return null; - } - public void addField(Field field) { - this.fields.add(field); - } + // check field type + List expectedClasses = expectedClassesFor(schema); - public FieldType getFieldType() { - return fieldType; + if (expectedClasses == null) { + throw new ConnectException("Invalid Java object for schema type " + schema.getFieldType() + + ": " + value.getClass() + + " for field: \"" + name + "\""); + } + + boolean foundMatch = false; + if (expectedClasses.size() == 1) { + foundMatch = expectedClasses.get(0).isInstance(value); + } else { + for (Class expectedClass : expectedClasses) { + if (expectedClass.isInstance(value)) { + foundMatch = true; + break; + } + } + } + + if (!foundMatch) { + throw new ConnectException("Invalid Java object for schema type " + schema.getFieldType() + + ": " + value.getClass() + + " for field: \"" + name + "\""); + } + + switch (schema.getFieldType()) { + case STRUCT: + Struct struct = (Struct) value; + if (!struct.schema().equals(schema)) { + throw new ConnectException("Struct schemas do not match."); + } + struct.validate(); + break; + case ARRAY: + List array = (List) value; + for (Object entry : array) { + validateValue(schema.valueSchema(), entry); + } + break; + case MAP: + Map map = (Map) value; + for (Map.Entry entry : map.entrySet()) { + validateValue(schema.keySchema(), entry.getKey()); + validateValue(schema.valueSchema(), entry.getValue()); + } + break; + } } - public void setFieldType(FieldType fieldType) { - this.fieldType = fieldType; + /** + * expected classes for + * @param schema + * @return + */ + private static List expectedClassesFor(Schema schema) { + return FieldType.SCHEMA_TYPE_CLASSES.get(schema.getFieldType()); } - @Override public String toString() { + @Override + public String toString() { return "Schema{" + - "name='" + name + '\'' + - ", fieldType=" + fieldType + - ", fields=" + fields + - '}'; + "name='" + name + '\'' + + ", version=" + version + + ", optional=" + optional + + ", defaultValue=" + defaultValue + + ", doc='" + doc + '\'' + + ", fieldType=" + fieldType + + ", fields=" + fields + + ", keySchema=" + keySchema + + ", valueSchema=" + valueSchema + + '}'; } } diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java b/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java index db353e4..786b6c6 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java +++ b/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java @@ -15,24 +15,62 @@ package io.openmessaging.connector.api.data; import io.openmessaging.connector.api.errors.ConnectException; +import io.openmessaging.connector.api.errors.SchemaBuilderException; + import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; +/** + * schema builder + */ public class SchemaBuilder { + private static final String TYPE_FIELD = "type"; + private static final String OPTIONAL_FIELD = "optional"; + private static final String DEFAULT_FIELD = "default"; + private static final String NAME_FIELD = "name"; + private static final String VERSION_FIELD = "version"; + private static final String DOC_FIELD = "doc"; /** * Name of the schema. */ private String name; + /** + * version + */ + private Integer version = null; + /** + * optional + */ + private Boolean optional = null; + /** + * default value + */ + private Object defaultValue = null; + /** + * Description schema + */ + private String doc = null; /** * Type of the field */ private FieldType type; + /** * Structure of the schema, contains a list of {@link Field} */ - private List fields; + private Map fields; + + /** + * map type + */ + private Schema keySchema; + private Schema valueSchema; + public SchemaBuilder(FieldType type) { if (null == type) { @@ -40,10 +78,19 @@ public SchemaBuilder(FieldType type) { } this.type = type; if (type == FieldType.STRUCT) { - fields = new ArrayList<>(); + /** + * ensure sequencing + */ + fields = new LinkedHashMap<>(); } } + + public boolean isOptional() { + return this.optional == null ? false : optional; + } + + /** * @return a new {@link FieldType#INT8} SchemaBuilder */ @@ -51,6 +98,13 @@ public static SchemaBuilder int8() { return new SchemaBuilder(FieldType.INT8); } + /** + * @return a new {@link FieldType#INT16} SchemaBuilder + */ + public static SchemaBuilder int16() { + return new SchemaBuilder(FieldType.INT16); + } + /** * @return a new {@link FieldType#INT32} SchemaBuilder */ @@ -109,24 +163,155 @@ public static SchemaBuilder struct() { return new SchemaBuilder(FieldType.STRUCT); } + + // Maps & Arrays + + /** + * @param valueSchema the schema for elements of the array + */ + public static SchemaBuilder array(Schema valueSchema) { + if (null == valueSchema) { + throw new SchemaBuilderException("valueSchema cannot be null."); + } + SchemaBuilder builder = new SchemaBuilder(FieldType.ARRAY); + builder.valueSchema = valueSchema; + return builder; + } + + /** + * @param keySchema the schema for keys in the map + * @param valueSchema the schema for values in the map + */ + public static SchemaBuilder map(Schema keySchema, Schema valueSchema) { + if (null == keySchema) { + throw new SchemaBuilderException("keySchema cannot be null."); + } + if (null == valueSchema) { + throw new SchemaBuilderException("valueSchema cannot be null."); + } + SchemaBuilder builder = new SchemaBuilder(FieldType.MAP); + builder.keySchema = keySchema; + builder.valueSchema = valueSchema; + return builder; + } + + /** + * Add a field to this struct schemaļ¼Œensure sequence + */ + public SchemaBuilder field(String fieldName, Schema fieldSchema) { + if (type != FieldType.STRUCT) { + throw new SchemaBuilderException("Cannot create fields on type " + type); + } + if (null == fieldName || fieldName.isEmpty()) { + throw new SchemaBuilderException("fieldName cannot be null."); + } + if (null == fieldSchema) { + throw new SchemaBuilderException("fieldSchema for field " + fieldName + " cannot be null."); + } + int fieldIndex = fields.size(); + if (fields.containsKey(fieldName)) { + throw new SchemaBuilderException("Cannot create field because of field name duplication " + fieldName); + } + fields.put(fieldName, new Field(fieldIndex, fieldName, fieldSchema)); + return this; + } + + + /** + * Get the list of fields for this Schema. Throws a DataException if this schema is not a struct. + */ + public List fields() { + if (type != FieldType.STRUCT) { + throw new ConnectException("Cannot list fields on non-struct type"); + } + return new ArrayList<>(fields.values()); + } + + /** + * get field by fieldName + * + * @param fieldName + * @return + */ + public Field field(String fieldName) { + if (type != FieldType.STRUCT) { + throw new ConnectException("Cannot look up fields on non-struct type"); + } + return fields.get(fieldName); + } + /** * Sets the schema name * * @param name schema name - * * @return the current SchemaBuilder instance */ public SchemaBuilder name(String name) { + checkCanSet(NAME_FIELD, this.name, name); this.name = name; return this; } + /** + * Set the version of this schema. Schema versions are integers which, if provided, must indicate which schema is + * newer and which is older by their ordering. + */ + public SchemaBuilder version(Integer version) { + checkCanSet(VERSION_FIELD, this.version, version); + this.version = version; + return this; + } + + /** + * Set the documentation for this schema. + */ + public SchemaBuilder doc(String doc) { + checkCanSet(DOC_FIELD, optional, true); + this.doc = doc; + return this; + } + + /** + * Set this schema as optional. + */ + public SchemaBuilder optional() { + checkCanSet(OPTIONAL_FIELD, optional, true); + optional = true; + return this; + } + + /** + * Set the default value for this schema. The value is validated against the schema type, throwing a + * {@link SchemaBuilderException} if it does not match. + */ + public SchemaBuilder defaultValue(Object value) { + checkCanSet(DEFAULT_FIELD, defaultValue, value); + checkNotNull(TYPE_FIELD, type, DEFAULT_FIELD); + defaultValue = value; + return this; + } + + + private static void checkCanSet(String fieldName, Object fieldVal, Object val) { + if (fieldVal != null && fieldVal != val) { + throw new ConnectException("Invalid SchemaBuilder call: " + fieldName + " has already been set."); + } + } + + + private static void checkNotNull(String fieldName, Object val, String fieldToSet) { + if (val == null) { + throw new SchemaBuilderException("Invalid SchemaBuilder call: " + fieldName + " must be specified to set " + fieldToSet); + } + } + /** * Build the Schema using the current settings * * @return the {@link Schema} */ public Schema build() { - return new Schema(name, type, fields); + return new Schema(name, type, isOptional(), defaultValue, version, doc, + fields == null ? null : Collections.unmodifiableList(new ArrayList<>(fields.values())), keySchema, valueSchema); } } diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/Struct.java b/connector/src/main/java/io/openmessaging/connector/api/data/Struct.java new file mode 100644 index 0000000..5710ef5 --- /dev/null +++ b/connector/src/main/java/io/openmessaging/connector/api/data/Struct.java @@ -0,0 +1,261 @@ + +package io.openmessaging.connector.api.data; + +import io.openmessaging.connector.api.errors.ConnectException; + +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +/** + * struct object model + */ +public class Struct { + + private final Schema schema; + private final Object[] values; + + /** + * construct + * @param schema + */ + public Struct(Schema schema) { + if (schema.getFieldType() != FieldType.STRUCT) { + throw new ConnectException("Not a struct schema: " + schema); + } + this.schema = schema; + this.values = new Object[schema.getFields().size()]; + } + + /** + * get schema + * @return + */ + public Schema schema() { + return schema; + } + + /** + * get data by fieldName + * @param fieldName + * @return + */ + public Object get(String fieldName) { + Field field = lookupField(fieldName); + return get(field); + } + + /** + * get data by Field object + * @param field + * @return + */ + public Object get(Field field) { + Object val = values[field.getIndex()]; + if (val == null && field.getSchema().defaultValue() != null) { + val = field.getSchema().defaultValue(); + } + return val; + } + + + public Object getWithoutDefault(String fieldName) { + Field field = lookupField(fieldName); + return values[field.getIndex()]; + } + + /** + * cast the result to a Byte. + * @param fieldName + * @return + */ + public Byte getInt8(String fieldName) { + return (Byte) getCheckType(fieldName, FieldType.INT8); + } + + /** + * cast the result to a Integer. + * @param fieldName + * @return + */ + public Integer getInt32(String fieldName) { + return (Integer) getCheckType(fieldName, FieldType.INT32); + } + + /** + * cast the result to a Long. + * @param fieldName + * @return + */ + public Long getInt64(String fieldName) { + return (Long) getCheckType(fieldName, FieldType.INT64); + } + + /** + * cast the result to a Float. + * @param fieldName + * @return + */ + public Float getFloat32(String fieldName) { + return (Float) getCheckType(fieldName, FieldType.FLOAT32); + } + + /** + * cast the result to a Double. + * @param fieldName + * @return + */ + public Double getFloat64(String fieldName) { + return (Double) getCheckType(fieldName, FieldType.FLOAT64); + } + + /** + * cast the result to a Boolean. + * @param fieldName + * @return + */ + public Boolean getBoolean(String fieldName) { + return (Boolean) getCheckType(fieldName, FieldType.BOOLEAN); + } + + /** + * cast the result to a String. + * @param fieldName + * @return + */ + public String getString(String fieldName) { + return (String) getCheckType(fieldName, FieldType.STRING); + } + + /** + * cast the result to a byte[]. + * @param fieldName + * @return + */ + public byte[] getBytes(String fieldName) { + Object bytes = getCheckType(fieldName, FieldType.BYTES); + if (bytes instanceof ByteBuffer) { + return ((ByteBuffer) bytes).array(); + } + return (byte[]) bytes; + } + + /** + * cast the result to a List. + * @param fieldName + * @param + * @return + */ + public List getArray(String fieldName) { + return (List) getCheckType(fieldName, FieldType.ARRAY); + } + + /** + * cast the result to a Map. + */ + public Map getMap(String fieldName) { + return (Map) getCheckType(fieldName, FieldType.MAP); + } + + /** + * cast the result to a Struct. + */ + public Struct getStruct(String fieldName) { + return (Struct) getCheckType(fieldName, FieldType.STRUCT); + } + + + /** + * Set value by fieldName + * @param fieldName + * @param value + * @return + */ + public Struct put(String fieldName, Object value) { + Field field = lookupField(fieldName); + return put(field, value); + } + + /** + * set data + * @param field + * @param value + * @return + */ + public Struct put(Field field, Object value) { + if (null == field) { + throw new ConnectException("field cannot be null."); + } + Schema.validateValue(field.getName(), field.getSchema(), value); + values[field.getIndex()] = value; + return this; + } + + + /** + * Validates that this struct has filled in all the necessary data with valid values + */ + public void validate() { + for (Field field : schema.getFields()) { + Schema fieldSchema = field.getSchema(); + Object value = values[field.getIndex()]; + if (value == null && (fieldSchema.isOptional() || fieldSchema.defaultValue() != null)) { + continue; + } + Schema.validateValue(field.getName(), fieldSchema, value); + } + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + Struct struct = (Struct) o; + return Objects.equals(schema, struct.schema) && + Arrays.deepEquals(values, struct.values); + } + + @Override + public int hashCode() { + return Objects.hash(schema, Arrays.deepHashCode(values)); + } + + private Field lookupField(String fieldName) { + Field field = schema.getField(fieldName); + if (field == null) { + throw new ConnectException(fieldName + " is not a valid field name"); + } + return field; + } + + private Object getCheckType(String fieldName, FieldType type) { + Field field = lookupField(fieldName); + if (field.getSchema().getFieldType() != type) { + throw new ConnectException("Field '" + fieldName + "' is not of type " + type); + } + return values[field.getIndex()]; + } + + @Override + public String toString() { + final StringBuilder sb = new StringBuilder("Struct{"); + boolean first = true; + for (int i = 0; i < values.length; i++) { + final Object value = values[i]; + if (value != null) { + final Field field = schema.getFields().get(i); + if (first) { + first = false; + } else { + sb.append(","); + } + sb.append(field.getName()).append("=").append(value); + } + } + return sb.append("}").toString(); + } + +} + diff --git a/connector/src/main/java/io/openmessaging/connector/api/errors/SchemaBuilderException.java b/connector/src/main/java/io/openmessaging/connector/api/errors/SchemaBuilderException.java new file mode 100644 index 0000000..5e88168 --- /dev/null +++ b/connector/src/main/java/io/openmessaging/connector/api/errors/SchemaBuilderException.java @@ -0,0 +1,28 @@ +/* + * Licensed 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. + */ +package io.openmessaging.connector.api.errors; + +public class SchemaBuilderException extends ConnectException { + public SchemaBuilderException(String s) { + super(s); + } + + public SchemaBuilderException(String s, Throwable throwable) { + super(s, throwable); + } + + public SchemaBuilderException(Throwable throwable) { + super(throwable); + } +} From c72659efc30c05c29b98d669e1cd1ab60811ac75 Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Tue, 24 May 2022 21:54:08 +0800 Subject: [PATCH 05/17] add offset storage writer #41 --- .../api/storage/OffsetStorageWriter.java | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) create mode 100644 connector/src/main/java/io/openmessaging/connector/api/storage/OffsetStorageWriter.java diff --git a/connector/src/main/java/io/openmessaging/connector/api/storage/OffsetStorageWriter.java b/connector/src/main/java/io/openmessaging/connector/api/storage/OffsetStorageWriter.java new file mode 100644 index 0000000..f41ef0b --- /dev/null +++ b/connector/src/main/java/io/openmessaging/connector/api/storage/OffsetStorageWriter.java @@ -0,0 +1,45 @@ +/* + * 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. + * + */ + +package io.openmessaging.connector.api.storage; + + +import io.openmessaging.connector.api.data.RecordOffset; +import io.openmessaging.connector.api.data.RecordPartition; + +import java.util.Map; + +/** + * position storage writer + */ +public interface OffsetStorageWriter { + + /** + * write offset + * @param partition + * @param position + */ + void writeOffset(RecordPartition partition, RecordOffset position); + + + /** + * write offsets + * @param positions + */ + void writeOffset(Map positions); +} From e6deeb8148f44e6e4c45fc852f8967b180ba8a34 Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Sat, 4 Jun 2022 12:12:08 +0800 Subject: [PATCH 06/17] add getter and setter method #41 --- .../connector/api/data/Schema.java | 62 +++++++++++++------ .../connector/api/data/SchemaBuilder.java | 2 +- .../connector/api/data/Struct.java | 29 +++++++-- 3 files changed, 69 insertions(+), 24 deletions(-) diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java b/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java index f4174ea..8858715 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java +++ b/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java @@ -123,36 +123,62 @@ public void addField(Field field) { this.fieldsByName.put(field.getName(), field); } - public Schema keySchema() { - return keySchema; + + public Integer getVersion() { + return version; } - public Schema valueSchema() { - return valueSchema; + + public void setVersion(Integer version) { + this.version = version; } public boolean isOptional() { return optional; } - public Object defaultValue(){ + public void setOptional(boolean optional) { + this.optional = optional; + } + + public Object getDefaultValue() { return defaultValue; } - /** - * Get the optional version of the schema. If a version is included, newer versions *must* be larger than older ones. - * @return the version of this schema - */ - public Integer version(){ - return version; + public void setDefaultValue(Object defaultValue) { + this.defaultValue = defaultValue; } - /** - * @return the documentation for this schema - */ - public String doc() { + public String getDoc() { return doc; } + public void setDoc(String doc) { + this.doc = doc; + } + + public Map getFieldsByName() { + return fieldsByName; + } + + public void setFieldsByName(Map fieldsByName) { + this.fieldsByName = fieldsByName; + } + + public Schema getKeySchema() { + return keySchema; + } + + public void setKeySchema(Schema keySchema) { + this.keySchema = keySchema; + } + + public Schema getValueSchema() { + return valueSchema; + } + + public void setValueSchema(Schema valueSchema) { + this.valueSchema = valueSchema; + } public String getName() { return name; @@ -239,14 +265,14 @@ public static void validateValue(String name, Schema schema, Object value) { case ARRAY: List array = (List) value; for (Object entry : array) { - validateValue(schema.valueSchema(), entry); + validateValue(schema.getValueSchema(), entry); } break; case MAP: Map map = (Map) value; for (Map.Entry entry : map.entrySet()) { - validateValue(schema.keySchema(), entry.getKey()); - validateValue(schema.valueSchema(), entry.getValue()); + validateValue(schema.getKeySchema(), entry.getKey()); + validateValue(schema.getValueSchema(), entry.getValue()); } break; } diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java b/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java index 786b6c6..4f0f468 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java +++ b/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java @@ -87,7 +87,7 @@ public SchemaBuilder(FieldType type) { public boolean isOptional() { - return this.optional == null ? false : optional; + return this.optional == null ? true : optional; } diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/Struct.java b/connector/src/main/java/io/openmessaging/connector/api/data/Struct.java index 5710ef5..a32d275 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/data/Struct.java +++ b/connector/src/main/java/io/openmessaging/connector/api/data/Struct.java @@ -14,8 +14,24 @@ */ public class Struct { - private final Schema schema; - private final Object[] values; + private Schema schema; + private Object[] values; + + public Schema getSchema() { + return schema; + } + + public void setSchema(Schema schema) { + this.schema = schema; + } + + public Object[] getValues() { + return values; + } + + public void setValues(Object[] values) { + this.values = values; + } /** * construct @@ -29,6 +45,9 @@ public Struct(Schema schema) { this.values = new Object[schema.getFields().size()]; } + + + /** * get schema * @return @@ -54,8 +73,8 @@ public Object get(String fieldName) { */ public Object get(Field field) { Object val = values[field.getIndex()]; - if (val == null && field.getSchema().defaultValue() != null) { - val = field.getSchema().defaultValue(); + if (val == null && field.getSchema().getDefaultValue() != null) { + val = field.getSchema().getDefaultValue(); } return val; } @@ -201,7 +220,7 @@ public void validate() { for (Field field : schema.getFields()) { Schema fieldSchema = field.getSchema(); Object value = values[field.getIndex()]; - if (value == null && (fieldSchema.isOptional() || fieldSchema.defaultValue() != null)) { + if (value == null && (fieldSchema.isOptional() || fieldSchema.getDefaultValue() != null)) { continue; } Schema.validateValue(field.getName(), fieldSchema, value); From aa9f315c846cc878ff8dfa673b9ad40777aa0df8 Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Sat, 4 Jun 2022 13:33:51 +0800 Subject: [PATCH 07/17] add SchemaAndValue #41 --- .../connector/api/data/SchemaAndValue.java | 64 +++++++++++++++++++ 1 file changed, 64 insertions(+) create mode 100644 connector/src/main/java/io/openmessaging/connector/api/data/SchemaAndValue.java diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/SchemaAndValue.java b/connector/src/main/java/io/openmessaging/connector/api/data/SchemaAndValue.java new file mode 100644 index 0000000..8b05436 --- /dev/null +++ b/connector/src/main/java/io/openmessaging/connector/api/data/SchemaAndValue.java @@ -0,0 +1,64 @@ +/* + * 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. + */ +package io.openmessaging.connector.api.data; + +import java.util.Objects; + +/** + * schema and value + */ +public class SchemaAndValue { + private final Schema schema; + private final Object value; + + public static final SchemaAndValue NULL = new SchemaAndValue(null, null); + + public SchemaAndValue(Schema schema, Object value) { + this.value = value; + this.schema = schema; + } + + public Schema schema() { + return schema; + } + + public Object value() { + return value; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + SchemaAndValue that = (SchemaAndValue) o; + return Objects.equals(schema, that.schema) && + Objects.equals(value, that.value); + } + + @Override + public int hashCode() { + return Objects.hash(schema, value); + } + + @Override + public String toString() { + return "SchemaAndValue{" + + "schema=" + schema + + ", value=" + value + + '}'; + } +} From 23ed62a077ce63cebe7b7561a9ee4eb0621f1e49 Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Sat, 4 Jun 2022 17:13:20 +0800 Subject: [PATCH 08/17] add logical type #41 --- .../connector/api/data/FieldType.java | 38 ----- .../connector/api/data/Schema.java | 149 ++++++++++++++---- .../connector/api/data/SchemaBuilder.java | 80 +++++++++- .../connector/api/data/logical/Date.java | 74 +++++++++ .../connector/api/data/logical/Decimal.java | 82 ++++++++++ .../connector/api/data/logical/Time.java | 81 ++++++++++ .../connector/api/data/logical/Timestamp.java | 59 +++++++ 7 files changed, 496 insertions(+), 67 deletions(-) create mode 100644 connector/src/main/java/io/openmessaging/connector/api/data/logical/Date.java create mode 100644 connector/src/main/java/io/openmessaging/connector/api/data/logical/Decimal.java create mode 100644 connector/src/main/java/io/openmessaging/connector/api/data/logical/Time.java create mode 100644 connector/src/main/java/io/openmessaging/connector/api/data/logical/Timestamp.java diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/FieldType.java b/connector/src/main/java/io/openmessaging/connector/api/data/FieldType.java index 0ff6467..71e169d 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/data/FieldType.java +++ b/connector/src/main/java/io/openmessaging/connector/api/data/FieldType.java @@ -14,14 +14,6 @@ package io.openmessaging.connector.api.data; -import java.nio.ByteBuffer; -import java.util.Arrays; -import java.util.Collections; -import java.util.EnumMap; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - /** * Define the field type. */ @@ -92,36 +84,6 @@ public enum FieldType { */ STRUCT; - /** - * Maps Schema.Types to a list of Java classes that can be used to represent them. - */ - public static final Map> SCHEMA_TYPE_CLASSES = new EnumMap<>(FieldType.class); - /** - * Maps the Java classes to the corresponding Schema.Type. - */ - public static final Map, FieldType> JAVA_CLASS_SCHEMA_TYPES = new HashMap<>(); - - static { - SCHEMA_TYPE_CLASSES.put(FieldType.INT8, Collections.singletonList((Class) Byte.class)); - SCHEMA_TYPE_CLASSES.put(FieldType.INT16, Collections.singletonList((Class) Short.class)); - SCHEMA_TYPE_CLASSES.put(FieldType.INT32, Collections.singletonList((Class) Integer.class)); - SCHEMA_TYPE_CLASSES.put(FieldType.INT64, Collections.singletonList((Class) Long.class)); - SCHEMA_TYPE_CLASSES.put(FieldType.FLOAT32, Collections.singletonList((Class) Float.class)); - SCHEMA_TYPE_CLASSES.put(FieldType.FLOAT64, Collections.singletonList((Class) Double.class)); - SCHEMA_TYPE_CLASSES.put(FieldType.BOOLEAN, Collections.singletonList((Class) Boolean.class)); - SCHEMA_TYPE_CLASSES.put(FieldType.STRING, Collections.singletonList((Class) String.class)); - SCHEMA_TYPE_CLASSES.put(FieldType.BYTES, Arrays.asList((Class) byte[].class, (Class) ByteBuffer.class)); - SCHEMA_TYPE_CLASSES.put(FieldType.ARRAY, Collections.singletonList((Class) List.class)); - SCHEMA_TYPE_CLASSES.put(FieldType.MAP, Collections.singletonList((Class) Map.class)); - SCHEMA_TYPE_CLASSES.put(FieldType.STRUCT, Collections.singletonList((Class) Struct.class)); - - for (Map.Entry> schemaClasses : SCHEMA_TYPE_CLASSES.entrySet()) { - for (Class schemaClass : schemaClasses.getValue()) { - JAVA_CLASS_SCHEMA_TYPES.put(schemaClass, schemaClasses.getKey()); - } - } - } - /** * Determine whether it is a basic type * diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java b/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java index 8858715..deac30d 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java +++ b/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java @@ -14,18 +14,71 @@ package io.openmessaging.connector.api.data; +import io.openmessaging.connector.api.data.logical.Date; +import io.openmessaging.connector.api.data.logical.Decimal; +import io.openmessaging.connector.api.data.logical.Time; +import io.openmessaging.connector.api.data.logical.Timestamp; import io.openmessaging.connector.api.errors.ConnectException; +import java.math.BigDecimal; +import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Collections; +import java.util.EnumMap; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; /** * Schema */ public class Schema { + /** + * Maps Schema.Types to a list of Java classes that can be used to represent them. + */ + public static final Map> SCHEMA_TYPE_CLASSES = new EnumMap<>(FieldType.class); + + /** + * Maps known logical types to a list of Java classes that can be used to represent them. + */ + private static final Map> LOGICAL_TYPE_CLASSES = new HashMap<>(); + + /** + * Maps the Java classes to the corresponding Schema.Type. + */ + public static final Map, FieldType> JAVA_CLASS_SCHEMA_TYPES = new HashMap<>(); + + static { + SCHEMA_TYPE_CLASSES.put(FieldType.INT8, Collections.singletonList((Class) Byte.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.INT16, Collections.singletonList((Class) Short.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.INT32, Collections.singletonList((Class) Integer.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.INT64, Collections.singletonList((Class) Long.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.FLOAT32, Collections.singletonList((Class) Float.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.FLOAT64, Collections.singletonList((Class) Double.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.BOOLEAN, Collections.singletonList((Class) Boolean.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.STRING, Collections.singletonList((Class) String.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.BYTES, Arrays.asList((Class) byte[].class, (Class) ByteBuffer.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.ARRAY, Collections.singletonList((Class) List.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.MAP, Collections.singletonList((Class) Map.class)); + SCHEMA_TYPE_CLASSES.put(FieldType.STRUCT, Collections.singletonList((Class) Struct.class)); + + for (Map.Entry> schemaClasses : SCHEMA_TYPE_CLASSES.entrySet()) { + for (Class schemaClass : schemaClasses.getValue()) { + JAVA_CLASS_SCHEMA_TYPES.put(schemaClass, schemaClasses.getKey()); + } + } + + LOGICAL_TYPE_CLASSES.put(Decimal.LOGICAL_NAME, Collections.singletonList((Class) BigDecimal.class)); + LOGICAL_TYPE_CLASSES.put(Date.LOGICAL_NAME, Collections.singletonList((Class) java.util.Date.class)); + LOGICAL_TYPE_CLASSES.put(Time.LOGICAL_NAME, Collections.singletonList((Class) java.util.Date.class)); + LOGICAL_TYPE_CLASSES.put(Timestamp.LOGICAL_NAME, Collections.singletonList((Class) java.util.Date.class)); + + } + + /** * Name of the schema. */ @@ -55,7 +108,7 @@ public class Schema { */ private List fields; private Map fieldsByName; - + private Map parameters; /** * map type */ @@ -65,13 +118,14 @@ public class Schema { /** * Construct a Schema. Most users should not construct schemas manually, preferring {@link SchemaBuilder} instead. */ - public Schema(String name,FieldType fieldType, boolean optional, Object defaultValue, Integer version, String doc, List fields,Schema keySchema, Schema valueSchema) { + public Schema(String name,FieldType fieldType, boolean optional, Object defaultValue, Integer version, String doc, List fields,Schema keySchema, Schema valueSchema, Map parameters) { this.fieldType = fieldType; this.optional = optional; this.defaultValue = defaultValue; this.name = name; this.version = version; this.doc = doc; + this.parameters = parameters; if (this.fieldType == FieldType.STRUCT) { this.fields = fields == null ? Collections.emptyList() : fields; this.fieldsByName = new HashMap<>(this.fields.size()); @@ -91,7 +145,7 @@ public Schema(String name,FieldType fieldType, boolean optional, Object defaultV * Construct a Schema. Most users should not construct schemas manually, preferring {@link SchemaBuilder} instead. */ public Schema(String name,FieldType fieldType, boolean optional, Object defaultValue, Integer version, List fields) { - this(name, fieldType, optional, defaultValue, version, null, fields, null, null); + this(name, fieldType, optional, defaultValue, version, null, fields, null, null,new ConcurrentHashMap<>()); } @@ -102,28 +156,7 @@ public Schema(String name, FieldType fieldType, List fields) { this(name, fieldType, false, null, null, fields); } - /** - * get field - * @param fieldName - * @return - */ - public Field getField(String fieldName) { - if (fieldsByName.containsKey(fieldName)){ - return fieldsByName.get(fieldName); - } - return null; - } - - /** - * add field - * @param field - */ - public void addField(Field field) { - this.fields.add(field); - this.fieldsByName.put(field.getName(), field); - } - - + // start getter and setter public Integer getVersion() { return version; } @@ -204,6 +237,37 @@ public void setFields(List fields) { this.fields = fields; } + public Map getParameters() { + return parameters; + } + + public void setParameters(Map parameters) { + this.parameters = parameters; + } + + // end getter and setter + + /** + * get field + * @param fieldName + * @return + */ + public Field getField(String fieldName) { + if (fieldsByName.containsKey(fieldName)){ + return fieldsByName.get(fieldName); + } + return null; + } + + /** + * add field + * @param field + */ + public void addField(Field field) { + this.fields.add(field); + this.fieldsByName.put(field.getName(), field); + } + /** * Validate that the value can be used for this schema */ @@ -217,7 +281,6 @@ public static void validateValue(Schema schema, Object value) { validateValue(null, schema, value); } public static void validateValue(String name, Schema schema, Object value) { - // check optional if (value == null) { if (!schema.isOptional()) { @@ -229,7 +292,6 @@ public static void validateValue(String name, Schema schema, Object value) { // check field type List expectedClasses = expectedClassesFor(schema); - if (expectedClasses == null) { throw new ConnectException("Invalid Java object for schema type " + schema.getFieldType() + ": " + value.getClass() @@ -284,9 +346,38 @@ public static void validateValue(String name, Schema schema, Object value) { * @return */ private static List expectedClassesFor(Schema schema) { - return FieldType.SCHEMA_TYPE_CLASSES.get(schema.getFieldType()); + List expectedClasses = LOGICAL_TYPE_CLASSES.get(schema.getName()); + if (expectedClasses == null) { + expectedClasses = SCHEMA_TYPE_CLASSES.getOrDefault(schema.getFieldType(), Collections.emptyList()); + } + return expectedClasses; + } + + + /** + * Get the type associated with the given class. + * @param klass + * @return + */ + public static FieldType schemaType(Class klass) { + synchronized (JAVA_CLASS_SCHEMA_TYPES) { + FieldType schemaType = JAVA_CLASS_SCHEMA_TYPES.get(klass); + if (Objects.nonNull(schemaType)) { + return schemaType; + } + for (Map.Entry, FieldType> entry : JAVA_CLASS_SCHEMA_TYPES.entrySet()) { + try { + klass.asSubclass(entry.getKey()); + JAVA_CLASS_SCHEMA_TYPES.put(klass, entry.getValue()); + return entry.getValue(); + } catch (ClassCastException e) { + } + } + } + return null; } + @Override public String toString() { return "Schema{" + @@ -297,6 +388,8 @@ public String toString() { ", doc='" + doc + '\'' + ", fieldType=" + fieldType + ", fields=" + fields + + ", fieldsByName=" + fieldsByName + + ", parameters=" + parameters + ", keySchema=" + keySchema + ", valueSchema=" + valueSchema + '}'; diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java b/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java index 4f0f468..1b92a3e 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java +++ b/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java @@ -14,6 +14,9 @@ package io.openmessaging.connector.api.data; +import io.openmessaging.connector.api.data.logical.Date; +import io.openmessaging.connector.api.data.logical.Decimal; +import io.openmessaging.connector.api.data.logical.Timestamp; import io.openmessaging.connector.api.errors.ConnectException; import io.openmessaging.connector.api.errors.SchemaBuilderException; @@ -64,6 +67,7 @@ public class SchemaBuilder { * Structure of the schema, contains a list of {@link Field} */ private Map fields; + private Map parameters; /** * map type @@ -195,6 +199,42 @@ public static SchemaBuilder map(Schema keySchema, Schema valueSchema) { return builder; } + + /** + * date + * @return + */ + public static SchemaBuilder date() { + return Date.builder(); + } + + /** + * time + * @return + */ + public static SchemaBuilder time() { + return Timestamp.builder(); + } + + /** + * timestamp + * @return + */ + public static SchemaBuilder timestamp() { + return Timestamp.builder(); + } + + /** + * decimal + * @param scale + * @return + */ + public static SchemaBuilder decimal(int scale) { + return Decimal.builder(scale); + } + + + /** * Add a field to this struct schemaļ¼Œensure sequence */ @@ -280,6 +320,44 @@ public SchemaBuilder optional() { return this; } + public Map parameters() { + return parameters == null ? null : Collections.unmodifiableMap(parameters); + } + + + /** + * Set a schema parameter + * @param propertyName + * @param propertyValue + * @return + */ + public SchemaBuilder parameter(String propertyName, String propertyValue) { + // Preserve order of insertion with a LinkedHashMap. This isn't strictly necessary, but is nice if logical types + // can print their properties in a consistent order. + if (parameters == null) { + parameters = new LinkedHashMap<>(); + } + parameters.put(propertyName, propertyValue); + return this; + } + + /** + * Set schema parameters + * @param props + * @return + */ + public SchemaBuilder parameters(Map props) { + // Avoid creating an empty set of properties so we never have an empty map + if (props.isEmpty()) { + return this; + } + if (parameters == null) { + parameters = new LinkedHashMap<>(); + } + parameters.putAll(props); + return this; + } + /** * Set the default value for this schema. The value is validated against the schema type, throwing a * {@link SchemaBuilderException} if it does not match. @@ -312,6 +390,6 @@ private static void checkNotNull(String fieldName, Object val, String fieldToSet */ public Schema build() { return new Schema(name, type, isOptional(), defaultValue, version, doc, - fields == null ? null : Collections.unmodifiableList(new ArrayList<>(fields.values())), keySchema, valueSchema); + fields == null ? null : Collections.unmodifiableList(new ArrayList<>(fields.values())), keySchema, valueSchema, parameters); } } diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/logical/Date.java b/connector/src/main/java/io/openmessaging/connector/api/data/logical/Date.java new file mode 100644 index 0000000..0908473 --- /dev/null +++ b/connector/src/main/java/io/openmessaging/connector/api/data/logical/Date.java @@ -0,0 +1,74 @@ +/* + * 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. + */ +package io.openmessaging.connector.api.data.logical; + +import io.openmessaging.connector.api.data.Schema; +import io.openmessaging.connector.api.data.SchemaBuilder; +import io.openmessaging.connector.api.errors.ConnectException; + +import java.util.Calendar; +import java.util.TimeZone; + +/** + * A date representing a calendar day with no time of day or timezone. + **/ +public class Date { + public static final String LOGICAL_NAME = "io.openmessaging.connector.api.data.logical.Date"; + + private static final long MILLIS_PER_DAY = 24 * 60 * 60 * 1000; + + private static final TimeZone UTC = TimeZone.getTimeZone("UTC"); + + + /** + * build schema + * @return + */ + public static SchemaBuilder builder() { + return SchemaBuilder.int32() + .name(LOGICAL_NAME) + .version(1); + } + + public static final Schema SCHEMA = builder().build(); + + /** + * Convert a value from its logical format (Date) to it's encoded format. + * @param value the logical value + * @return the encoded value + */ + public static int fromLogical(Schema schema, java.util.Date value) { + if (!(LOGICAL_NAME.equals(schema.getName()))) { + throw new ConnectException("Requested conversion of Date object but the schema does not match."); + } + Calendar calendar = Calendar.getInstance(UTC); + calendar.setTime(value); + if (calendar.get(Calendar.HOUR_OF_DAY) != 0 || calendar.get(Calendar.MINUTE) != 0 || + calendar.get(Calendar.SECOND) != 0 || calendar.get(Calendar.MILLISECOND) != 0) { + throw new ConnectException("Kafka Connect Date type should not have any time fields set to non-zero values."); + } + long unixMillis = calendar.getTimeInMillis(); + return (int) (unixMillis / MILLIS_PER_DAY); + } + + public static java.util.Date toLogical(Schema schema, int value) { + if (!(LOGICAL_NAME.equals(schema.getName()))) { + throw new ConnectException("Requested conversion of Date object but the schema does not match."); + } + return new java.util.Date(value * MILLIS_PER_DAY); + } +} diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/logical/Decimal.java b/connector/src/main/java/io/openmessaging/connector/api/data/logical/Decimal.java new file mode 100644 index 0000000..2781710 --- /dev/null +++ b/connector/src/main/java/io/openmessaging/connector/api/data/logical/Decimal.java @@ -0,0 +1,82 @@ +/* + * 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. + */ +package io.openmessaging.connector.api.data.logical; + +import io.openmessaging.connector.api.data.Schema; +import io.openmessaging.connector.api.data.SchemaBuilder; +import io.openmessaging.connector.api.errors.ConnectException; + +import java.math.BigDecimal; +import java.math.BigInteger; + +/** + * decimal + */ +public class Decimal { + public static final String LOGICAL_NAME = "io.openmessaging.connector.api.data.logical.Decimal"; + public static final String SCALE_FIELD = "scale"; + + /** + *schema builder + * @param scale + * @return + */ + public static SchemaBuilder builder(int scale) { + return SchemaBuilder.bytes() + .name(LOGICAL_NAME) + .parameter(SCALE_FIELD, Integer.toString(scale)) + .version(1); + } + + /** + * get schema + * @param scale + * @return + */ + public static Schema schema(int scale){ + return builder(scale).build(); + } + + /** + * Convert a value from its logical format (BigDecimal) to it's encoded format. + * @param schema + * @param value + * @return + */ + public static byte[] fromLogical(Schema schema, BigDecimal value) { + if (value.scale() != scale(schema)) { + throw new ConnectException("BigDecimal has mismatching scale value for given Decimal schema"); + } + return value.unscaledValue().toByteArray(); + } + + public static BigDecimal toLogical(Schema schema, byte[] value) { + return new BigDecimal(new BigInteger(value), scale(schema)); + } + + private static int scale(Schema schema) { + String scaleString = schema.getParameters().get(SCALE_FIELD); + if (scaleString == null) { + throw new ConnectException("Invalid Decimal schema: scale parameter not found."); + } + try { + return Integer.parseInt(scaleString); + } catch (NumberFormatException e) { + throw new ConnectException("Invalid scale parameter found in Decimal schema: ", e); + } + } +} diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/logical/Time.java b/connector/src/main/java/io/openmessaging/connector/api/data/logical/Time.java new file mode 100644 index 0000000..f52158c --- /dev/null +++ b/connector/src/main/java/io/openmessaging/connector/api/data/logical/Time.java @@ -0,0 +1,81 @@ +/* + * 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. + */ +package io.openmessaging.connector.api.data.logical; + +import io.openmessaging.connector.api.data.Schema; +import io.openmessaging.connector.api.data.SchemaBuilder; +import io.openmessaging.connector.api.errors.ConnectException; + +import java.util.Calendar; +import java.util.TimeZone; + + +/** + * A time representing a specific point in a day, not tied to any specific date. The corresponding Java type is a + * java.util.Date where only hours, minutes, seconds, and milliseconds can be non-zero. This effectively makes it a + * point in time during the first day after the Unix epoch. The underlying representation is an integer + * representing the number of milliseconds after midnight. + */ +public class Time { + public static final String LOGICAL_NAME = "io.openmessaging.connector.api.data.logical.Time"; + + private static final long MILLIS_PER_DAY = 24 * 60 * 60 * 1000; + + private static final TimeZone UTC = TimeZone.getTimeZone("UTC"); + + /** + * Returns a SchemaBuilder for a Time. By returning a SchemaBuilder you can override additional schema settings such + * as required/optional, default value, and documentation. + * @return a SchemaBuilder + */ + private static SchemaBuilder builder() { + return SchemaBuilder.int32() + .name(LOGICAL_NAME) + .version(1); + } + + public static final Schema SCHEMA = builder().build(); + + /** + * Convert a value from its logical format (Time) to it's encoded format. + * @param schema + * @param value + * @return + */ + public static int fromLogical(Schema schema, java.util.Date value) { + if (!(LOGICAL_NAME.equals(schema.getName()))) { + throw new ConnectException("Requested conversion of Time object but the schema does not match."); + } + Calendar calendar = Calendar.getInstance(UTC); + calendar.setTime(value); + long unixMillis = calendar.getTimeInMillis(); + if (unixMillis < 0 || unixMillis > MILLIS_PER_DAY) { + throw new ConnectException("Kafka Connect Time type should not have any date fields set to non-zero values."); + } + return (int) unixMillis; + } + + public static java.util.Date toLogical(Schema schema, int value) { + if (!(LOGICAL_NAME.equals(schema.getName()))) { + throw new ConnectException("Requested conversion of Date object but the schema does not match."); + } + if (value < 0 || value > MILLIS_PER_DAY) { + throw new ConnectException("Time values must use number of milliseconds greater than 0 and less than 86400000"); + } + return new java.util.Date(value); + } +} diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/logical/Timestamp.java b/connector/src/main/java/io/openmessaging/connector/api/data/logical/Timestamp.java new file mode 100644 index 0000000..7f00ce6 --- /dev/null +++ b/connector/src/main/java/io/openmessaging/connector/api/data/logical/Timestamp.java @@ -0,0 +1,59 @@ +/* + * 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. + */ +package io.openmessaging.connector.api.data.logical; + +import io.openmessaging.connector.api.data.Schema; +import io.openmessaging.connector.api.data.SchemaBuilder; +import io.openmessaging.connector.api.errors.ConnectException; + +/** + * A timestamp representing an absolute time, without timezone information. The corresponding Java type is a + * java.util.Date. The underlying representation is a long representing the number of milliseconds since Unix epoch. + */ +public class Timestamp { + public static final String LOGICAL_NAME = "io.openmessaging.connector.api.data.logical.Timestamp"; + + + /** + * build schema + * @return + */ + public static SchemaBuilder builder() { + return SchemaBuilder.int64() + .name(LOGICAL_NAME) + .version(1); + } + + public static final Schema SCHEMA = builder().build(); + + /** + * Convert a value from its logical format (Date) to it's encoded format. + */ + public static long fromLogical(Schema schema, java.util.Date value) { + if (!(LOGICAL_NAME.equals(schema.getName()))) { + throw new ConnectException("Requested conversion of Timestamp object but the schema does not match."); + } + return value.getTime(); + } + + public static java.util.Date toLogical(Schema schema, long value) { + if (!(LOGICAL_NAME.equals(schema.getName()))) { + throw new ConnectException("Requested conversion of Timestamp object but the schema does not match."); + } + return new java.util.Date(value); + } +} From 26fe155363e5710e3ba33829a292438965ad742c Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Mon, 6 Jun 2022 16:02:50 +0800 Subject: [PATCH 09/17] Schemabuilder add required method --- .../openmessaging/connector/api/data/SchemaBuilder.java | 9 +++++++++ .../openmessaging/connector/api/data/logical/Time.java | 2 +- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java b/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java index 1b92a3e..d29738c 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java +++ b/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java @@ -320,6 +320,15 @@ public SchemaBuilder optional() { return this; } + /** + * Set this schema as required. + */ + public SchemaBuilder required() { + checkCanSet(OPTIONAL_FIELD, optional, false); + optional = false; + return this; + } + public Map parameters() { return parameters == null ? null : Collections.unmodifiableMap(parameters); } diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/logical/Time.java b/connector/src/main/java/io/openmessaging/connector/api/data/logical/Time.java index f52158c..6638d8f 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/data/logical/Time.java +++ b/connector/src/main/java/io/openmessaging/connector/api/data/logical/Time.java @@ -42,7 +42,7 @@ public class Time { * as required/optional, default value, and documentation. * @return a SchemaBuilder */ - private static SchemaBuilder builder() { + public static SchemaBuilder builder() { return SchemaBuilder.int32() .name(LOGICAL_NAME) .version(1); From 3e6f1e6a4912fcdfb8c7b356a758e5f04bdcef8c Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Mon, 6 Jun 2022 16:12:20 +0800 Subject: [PATCH 10/17] schema add hashCode and equals method --- .../io/openmessaging/connector/api/data/Schema.java | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java b/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java index deac30d..3a24eca 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java +++ b/connector/src/main/java/io/openmessaging/connector/api/data/Schema.java @@ -377,6 +377,18 @@ public static FieldType schemaType(Class klass) { return null; } + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (!(o instanceof Schema)) return false; + Schema schema = (Schema) o; + return isOptional() == schema.isOptional() && Objects.equals(getName(), schema.getName()) && Objects.equals(getVersion(), schema.getVersion()) && Objects.equals(getDefaultValue(), schema.getDefaultValue()) && Objects.equals(getDoc(), schema.getDoc()) && getFieldType() == schema.getFieldType() && Objects.equals(getFields(), schema.getFields()) && Objects.equals(getParameters(), schema.getParameters()) && Objects.equals(getKeySchema(), schema.getKeySchema()) && Objects.equals(getValueSchema(), schema.getValueSchema()); + } + + @Override + public int hashCode() { + return Objects.hash(getName(), getVersion(), isOptional(), getDefaultValue(), getDoc(), getFieldType(), getFields(), getParameters(), getKeySchema(), getValueSchema()); + } @Override public String toString() { From 14765cb460a61de6cc801f8a9bd1b94e821ca55f Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Mon, 6 Jun 2022 21:15:29 +0800 Subject: [PATCH 11/17] fixed doc method --- .../java/io/openmessaging/connector/api/data/SchemaBuilder.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java b/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java index d29738c..a47fc6c 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java +++ b/connector/src/main/java/io/openmessaging/connector/api/data/SchemaBuilder.java @@ -306,7 +306,7 @@ public SchemaBuilder version(Integer version) { * Set the documentation for this schema. */ public SchemaBuilder doc(String doc) { - checkCanSet(DOC_FIELD, optional, true); + checkCanSet(DOC_FIELD, this.doc, doc); this.doc = doc; return this; } From a6b3c79a2c916ba92f481a876db3232956ffd0f4 Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Tue, 7 Jun 2022 12:32:31 +0800 Subject: [PATCH 12/17] Field add equals and hashcode method --- .../openmessaging/connector/api/data/Field.java | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/Field.java b/connector/src/main/java/io/openmessaging/connector/api/data/Field.java index 73d53f7..ca8bfe6 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/data/Field.java +++ b/connector/src/main/java/io/openmessaging/connector/api/data/Field.java @@ -14,6 +14,8 @@ package io.openmessaging.connector.api.data; +import java.util.Objects; + /** * Filed of the schema. */ @@ -62,6 +64,20 @@ public void setSchema(Schema schema) { this.schema = schema; } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (!(o instanceof Field)) return false; + Field field = (Field) o; + return getIndex() == field.getIndex() && Objects.equals(getName(), field.getName()) && Objects.equals(getSchema(), field.getSchema()); + } + + @Override + public int hashCode() { + return Objects.hash(getIndex(), getName(), getSchema()); + } + @Override public String toString() { return "Field{" + "index=" + index + From 427e009dd0d9a950bc5a20f5d4216b840bb1ad1b Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Wed, 8 Jun 2022 10:53:10 +0800 Subject: [PATCH 13/17] optimize api #85 --- .../api/component/connector/Connector.java | 21 ++++++----- .../connector/api/component/task/Task.java | 16 ++++----- .../component/task/sink/SinkConnector.java | 3 ++ .../api/component/task/sink/SinkTask.java | 29 ++------------- .../component/task/sink/SinkTaskContext.java | 3 ++ .../task/source/SourceConnector.java | 3 ++ .../api/component/task/source/SourceTask.java | 35 ++----------------- 7 files changed, 30 insertions(+), 80 deletions(-) diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/connector/Connector.java b/connector/src/main/java/io/openmessaging/connector/api/component/connector/Connector.java index d9272a5..d444bf1 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/connector/Connector.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/connector/Connector.java @@ -23,20 +23,11 @@ public abstract class Connector implements Component { protected ConnectorContext connectorContext; - @Override public void init(ConnectorContext connectorContext) { + @Override + public void init(ConnectorContext connectorContext) { this.connectorContext = connectorContext; } - /** - * Pause the connector. - */ - public abstract void pause(); - - /** - * Resume the connector. - */ - public abstract void resume(); - /** * Returns a set of configurations for Tasks based on the current configuration, * producing at most count configurations. @@ -53,4 +44,12 @@ public abstract class Connector implements Component { */ public abstract Class taskClass(); + /** + * Should invoke before start the connector. + * @param config component config + */ + @Override + public void validate(KeyValue config){ + + } } diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/task/Task.java b/connector/src/main/java/io/openmessaging/connector/api/component/task/Task.java index 5c34f59..7b83931 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/task/Task.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/task/Task.java @@ -14,21 +14,17 @@ package io.openmessaging.connector.api.component.task; +import io.openmessaging.KeyValue; import io.openmessaging.connector.api.component.Component; import io.openmessaging.connector.api.component.ComponentContext; public interface Task extends Component { /** - * Pause the task. + * Should invoke before start the connector. + * + * @param config component config */ - @Deprecated - void pause(); - - /** - * Resume the task. - */ - @Deprecated - void resume(); - + @Override + default void validate(KeyValue config) {} } diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkConnector.java b/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkConnector.java index c1f46af..a1a5f60 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkConnector.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkConnector.java @@ -16,6 +16,9 @@ import io.openmessaging.connector.api.component.connector.Connector; +/** + * sink connector + */ public abstract class SinkConnector extends Connector { public SinkConnector() { diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java b/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java index 9fc8f7a..9e722dc 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java @@ -14,7 +14,6 @@ package io.openmessaging.connector.api.component.task.sink; -import io.openmessaging.KeyValue; import io.openmessaging.connector.api.component.task.Task; import io.openmessaging.connector.api.data.ConnectRecord; import io.openmessaging.connector.api.data.RecordOffset; @@ -23,6 +22,9 @@ import java.util.List; import java.util.Map; +/** + * sink task + */ public abstract class SinkTask implements Task { protected SinkTaskContext sinkTaskContext; @@ -58,29 +60,4 @@ public Map preCommit(Map { protected SourceTaskContext sourceTaskContext; - @Override public void init(SourceTaskContext sourceTaskContext) { + @Override + public void init(SourceTaskContext sourceTaskContext) { this.sourceTaskContext = sourceTaskContext; } @@ -56,35 +56,4 @@ public void commit(final List connectRecords) throws InterruptedE * function. */ public void commit() { } - - /** - * Get source task context. - * @return source task context - */ - @Deprecated - public SourceTaskContext getContext() { - return sourceTaskContext; - } - - @Override - public void validate(KeyValue config){ - - } - - /** - * Pause the task. - */ - @Override - public void pause() { - - } - - /** - * Resume the task. - */ - @Override - public void resume() { - - } - } From c4b1836e6be105124925fece4d49b516ac6a0b53 Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Thu, 9 Jun 2022 19:57:43 +0800 Subject: [PATCH 14/17] Optimize transform api #45 --- .../connector/api/component/Transform.java | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/Transform.java b/connector/src/main/java/io/openmessaging/connector/api/component/Transform.java index d912b25..170e62d 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/Transform.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/Transform.java @@ -14,6 +14,7 @@ package io.openmessaging.connector.api.component; +import io.openmessaging.KeyValue; import io.openmessaging.connector.api.data.ConnectRecord; /** @@ -22,7 +23,23 @@ * @version OMS 0.1.0 * @since OMS 0.1.0 */ -public interface Transform extends Component { +public interface Transform extends AutoCloseable { + /** + * config + * @param config + */ + void config(KeyValue config); + /** + * transform record + * @param record + * @return + */ R doTransform(R record); + + /** + * close + */ + @Override + void close(); } From c76d2fb339a66d39694197be61c5814ca455d4c0 Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Sat, 18 Jun 2022 11:52:43 +0800 Subject: [PATCH 15/17] Optimize transform api and add RecordConverter --- .../connector/api/component/Component.java | 9 +-- .../connector/api/component/Transform.java | 18 ++--- .../api/component/connector/Connector.java | 8 +- .../connector/api/component/task/Task.java | 9 ++- .../api/component/task/sink/SinkTask.java | 5 +- .../api/component/task/source/SourceTask.java | 4 + .../connector/api/data/RecordConverter.java | 81 +++++++++++++++++++ 7 files changed, 111 insertions(+), 23 deletions(-) create mode 100644 connector/src/main/java/io/openmessaging/connector/api/data/RecordConverter.java diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/Component.java b/connector/src/main/java/io/openmessaging/connector/api/component/Component.java index 0349c2f..8ee512d 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/Component.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/Component.java @@ -16,7 +16,7 @@ import io.openmessaging.KeyValue; -public interface Component { +public interface Component { /** * Should invoke before start the connector. @@ -25,13 +25,6 @@ public interface Component { */ void validate(KeyValue config); - /** - * Init the component - * - * @param context component context - */ - void init(R context); - /** * Start the component * diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/Transform.java b/connector/src/main/java/io/openmessaging/connector/api/component/Transform.java index 170e62d..c48cf70 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/Transform.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/Transform.java @@ -23,23 +23,21 @@ * @version OMS 0.1.0 * @since OMS 0.1.0 */ -public interface Transform extends AutoCloseable { +public interface Transform extends Component { + /** - * config - * @param config + * Should invoke before start the connector. + * + * @param config component config */ - void config(KeyValue config); + @Override + default void validate(KeyValue config){ + } /** * transform record * @param record * @return */ R doTransform(R record); - - /** - * close - */ - @Override - void close(); } diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/connector/Connector.java b/connector/src/main/java/io/openmessaging/connector/api/component/connector/Connector.java index d444bf1..cdde4a1 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/connector/Connector.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/connector/Connector.java @@ -16,18 +16,22 @@ import io.openmessaging.KeyValue; import io.openmessaging.connector.api.component.Component; +import io.openmessaging.connector.api.component.ComponentContext; import io.openmessaging.connector.api.component.task.Task; import java.util.List; -public abstract class Connector implements Component { +public abstract class Connector implements Component { protected ConnectorContext connectorContext; - @Override public void init(ConnectorContext connectorContext) { this.connectorContext = connectorContext; } + protected ConnectorContext getConnectorContext() { + return connectorContext; + } + /** * Returns a set of configurations for Tasks based on the current configuration, * producing at most count configurations. diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/task/Task.java b/connector/src/main/java/io/openmessaging/connector/api/component/task/Task.java index 7b83931..73fbf45 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/task/Task.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/task/Task.java @@ -18,7 +18,7 @@ import io.openmessaging.connector.api.component.Component; import io.openmessaging.connector.api.component.ComponentContext; -public interface Task extends Component { +public interface Task extends Component{ /** * Should invoke before start the connector. @@ -27,4 +27,11 @@ public interface Task extends Component { */ @Override default void validate(KeyValue config) {} + + /** + * Init the component + * + * @param context component context + */ + void init(R context); } diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java b/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java index 9e722dc..ea8d8ff 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/task/sink/SinkTask.java @@ -23,13 +23,14 @@ import java.util.Map; /** - * sink task + * The sink task API definition is used to define the logic for data writing */ public abstract class SinkTask implements Task { protected SinkTaskContext sinkTaskContext; - @Override public void init(SinkTaskContext sinkTaskContext) { + @Override + public void init(SinkTaskContext sinkTaskContext) { this.sinkTaskContext = sinkTaskContext; } diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java b/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java index 4a91049..4d69c56 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java @@ -18,9 +18,13 @@ import io.openmessaging.connector.api.data.ConnectRecord; import java.util.List; +/** + * The source task API definition is used to define the logic for data pulling + */ public abstract class SourceTask implements Task { protected SourceTaskContext sourceTaskContext; + @Override public void init(SourceTaskContext sourceTaskContext) { this.sourceTaskContext = sourceTaskContext; diff --git a/connector/src/main/java/io/openmessaging/connector/api/data/RecordConverter.java b/connector/src/main/java/io/openmessaging/connector/api/data/RecordConverter.java new file mode 100644 index 0000000..6912492 --- /dev/null +++ b/connector/src/main/java/io/openmessaging/connector/api/data/RecordConverter.java @@ -0,0 +1,81 @@ +/* + * 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. + */ +package io.openmessaging.connector.api.data; + +import io.openmessaging.KeyValue; + +import java.util.Map; + +/** + * The topic parameter is mainly a compatible schema registry + * abstract converter + */ +public interface RecordConverter { + + + /** + * Config is used for parameter passing in the conversion process + * @param configs + */ + default void configure(Map configs) { + + } + + /** + * Convert ConnectRecord to byte[] + * @param topic the topic associated with the data + * @param schema record schema + * @param value record value + * @return + */ + byte[] fromConnectData(String topic, Schema schema, Object value); + + + /** + * The provided subject and extension may be used in the record as needed. + * @param topic the topic associated with the data + * @param extensions + * @param schema rocketmq connect record schema + * @param value rocketmq connect record value + * @return + */ + default byte[] fromConnectData(String topic, KeyValue extensions, Schema schema, Object value) { + return fromConnectData(topic, schema, value); + } + + /** + * Convert a byte[] object to a Rocketmq Connect data object. + * + * @param topic the topic associated with the data + * @param value the value to convert + * @return an object containing the {@link Schema} and the converted value + */ + SchemaAndValue toConnectData(String topic, byte[] value); + + + /** + * The provided subject and extension may be used in the record as needed. + * @param topic the topic associated with the data + * @param extensions transform property + * @param value + * @return + */ + default SchemaAndValue toConnectData(String topic, KeyValue extensions, byte[] value) { + return toConnectData(topic, value); + } + +} From c2205c3a5e791ab012fb3f3710fc38a279c33098 Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Mon, 4 Jul 2022 18:17:17 +0800 Subject: [PATCH 16/17] Optimize the source task commit method #51 --- .../api/component/task/source/SourceTask.java | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java b/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java index 4d69c56..29c9ea5 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java @@ -16,7 +16,9 @@ import io.openmessaging.connector.api.component.task.Task; import io.openmessaging.connector.api.data.ConnectRecord; + import java.util.List; +import java.util.Map; /** * The source task API definition is used to define the logic for data pulling @@ -38,6 +40,15 @@ public void init(SourceTaskContext sourceTaskContext) { */ public abstract List poll() throws InterruptedException; + /** + * commit record + * @param record + * @param metadata + */ + public void commit(final ConnectRecord record, Map metadata) { + commit(record); + } + /** *

* Commit an individual {@link ConnectRecord} when the callback from the producer client is received. @@ -48,11 +59,9 @@ public void init(SourceTaskContext sourceTaskContext) { * in their own system. *

* - * @throws InterruptedException task thread interupt exception - * @param connectRecords connect records + * @param record connect record */ - public void commit(final List connectRecords) throws InterruptedException { - commit(); + public void commit(final ConnectRecord record) { } /** From 5d7f13f8ef052ded02f9484182b68b9668a07890 Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Wed, 13 Jul 2022 20:49:30 +0800 Subject: [PATCH 17/17] add batch commit #51 --- .../connector/api/component/task/source/SourceTask.java | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java b/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java index 29c9ea5..2ea4ce1 100644 --- a/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java +++ b/connector/src/main/java/io/openmessaging/connector/api/component/task/source/SourceTask.java @@ -40,6 +40,14 @@ public void init(SourceTaskContext sourceTaskContext) { */ public abstract List poll() throws InterruptedException; + + /** + * batch commit + * @param records + * @param metadata + */ + public void commit(final List records, Map metadata) { + } /** * commit record * @param record