add flink-parse-avro demo

src/flink-examples/flink-examples-security/flink-parse-avro/pom.xml

@@ -0,0 +1,117 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>com.huawei.mrs</groupId>
+    <artifactId>FlinkKafkaAvroParser</artifactId>
+    <version>1.0</version>
+
+    <properties>
+        <flink.version>1.12.2-hw-ei-312005</flink.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-core</artifactId>
+            <version>${flink.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-hadoop-2</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-zookeeper-3</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-java</artifactId>
+            <version>${flink.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-hadoop-2</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-zookeeper-3</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java_2.11</artifactId>
+            <version>${flink.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-hadoop-2</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-zookeeper-3</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-kafka_2.11</artifactId>
+            <version>${flink.version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <configuration>
+                    <archive>
+                        <manifest>
+                            <addClasspath>true</addClasspath>
+                            <!--指定 resources/lib 目录-->
+                            <classpathPrefix>lib/</classpathPrefix>
+                            <!--应用的主入口类-->
+<!--                            <mainClass>com.huawei.bigdata.kafka.example.Consumer</mainClass>-->
+                        </manifest>
+                    </archive>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-dependency-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>copy</id>
+                        <phase>compile</phase>
+                        <goals>
+                            <!--将 resources/lib 目录所有 Jar 包打进最终的依赖中-->
+                            <goal>copy-dependencies</goal>
+                        </goals>
+                        <configuration>
+                            <!--将 resources/lib 目录所有 Jar 包一并拷贝到输出目录的 lib 目录下-->
+                            <outputDirectory>
+                                ${project.build.directory}/lib
+                            </outputDirectory>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>1.8</source>
+                    <target>1.8</target>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+
+</project>

src/flink-examples/flink-examples-security/flink-parse-avro/src/main/java/com/huawei/bigdata/flink/avro/parser/Avro2Json.java

@@ -0,0 +1,230 @@
+package com.huawei.bigdata.flink.avro.parser;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.huawei.hwclouds.drs.avro.Record;
+import org.apache.commons.lang.StringUtils;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeHint;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
+import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
+import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
+import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class Avro2Json {
+    public static final String TOPIC = "topic";
+
+    public static final String DRS_VERSION = "version";
+    public static final String DRS_SEQ_NO = "seqno";
+    public static final String DRS_SHARD_ID = "shardId";
+    public static final String DRS_EVENT_ID = "eventid";
+    public static final String DRS_UPDATE_TIMESTAMP = "updateTimestamp";
+    public static final String DRS_TABLE_NAME = "tableName";
+    public static final String DRS_OPERATION = "operation";
+    public static final String DRS_COLUMN_COUNT = "columnCount";
+    public static final String DRS_FIELDS = "fields";
+    public static final String DRS_BEFORE_IMAGES = "beforeImages";
+    public static final String DRS_AFTER_IMAGES = "afterImages";
+
+    public static final String DEBEZIUM_TS_MS = "ts_ms";
+    public static final String DEBEZIUM_OP = "op";
+    public static final String DEBEZIUM_SOURCE = "source";
+    public static final String DEBEZIUM_BEFORE = "before";
+    public static final String DEBEZIUM_AFTER = "after";
+    public static final ObjectMapper JSON_MAPPER = new ObjectMapper();
+
+    public static void main(String[] args) throws Exception {
+
+        ParameterTool paraTool = ParameterTool.fromArgs(args);
+        Properties props = paraTool.getProperties();
+        String avroTopic = props.getProperty("avro_topic", "default_avro_topic");
+        String jsonTopic = props.getProperty("json_topic", "default_json_topic");
+        int parallelism = Integer.parseInt(props.getProperty("partition_num", "1"));
+
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+        env.setParallelism(parallelism);
+
+        DataStream<String> messageStream =
+                env.addSource(
+                        new FlinkKafkaConsumer<>(
+                                avroTopic, new KafkaDeserializationSchema<String>() {
+                            @Override
+                            public boolean isEndOfStream(String s) {
+                                return false;
+                            }
+
+                            @Override
+                            public String deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) {
+                                String str = "";
+                                    try {
+                                        if (consumerRecord != null && consumerRecord.key() != null && consumerRecord.value() != null) {
+                                            Object rec = Record.getDecoder().decode(consumerRecord.value());
+                                            str = rec.toString();
+                                        }
+                                    } catch (IOException exception) {
+                                        exception.printStackTrace();
+                                    }
+                                return str;
+                            }
+
+                            @Override
+                            public TypeInformation<String> getProducedType() {
+                                return TypeInformation.of(new TypeHint<String>(){});
+                            }
+                        }, props));
+
+//        messageStream.filter(s -> !s.isEmpty()).map(new MapFunction<String, Map<String, Object>>() {
+//            @Override
+//            public Map<String, Object> map(String jsonStr) throws Exception {
+//                Map<String, Object> map = new HashMap<>();
+//                Map<String, Object> mapAll = mapper.readValue(jsonStr, new TypeReference<HashMap<String, Object>>() {
+//                });
+//                map.put(TABLE_NAME, StringUtils.remove((String) mapAll.get(TABLE_NAME), '\"'));
+//                map.put(UPDATE_TIMESTAMP, mapAll.get(UPDATE_TIMESTAMP));
+//                map.put(OPERATION, mapAll.get(OPERATION));
+//                map.put(COLUMN_NAMES, mapAll.get(FIELDS));
+//                List<Object> list = (List<Object>) mapAll.get(AFTER_IMAGES);
+//                List<Object> newList = list.stream().map(obj -> {
+//                    if (obj == null) {
+//                        return new HashMap<String, String>() {{
+//                            put("value", null);
+//                        }};
+//                    } else {
+//                        return obj;
+//                    }
+//                }).collect(Collectors.toList());
+//                map.put(COLUMN_VALUES, newList);
+
+        messageStream.filter(str -> !str.isEmpty()).map(new MapFunction<String, Map<String, Object>>() {
+            @Override
+            public Map<String, Object> map(String jsonStr) throws Exception {
+                Map<String, Object> drsJsonMap = JSON_MAPPER.readValue(jsonStr, HashMap.class);
+                String version = String.valueOf(drsJsonMap.get(DRS_VERSION));
+                String seqno = String.valueOf(drsJsonMap.get(DRS_SEQ_NO));
+                String shardId = String.valueOf(drsJsonMap.get(DRS_SHARD_ID));
+                String eventId = String.valueOf(drsJsonMap.get(DRS_EVENT_ID));
+                String updateTimestamp = String.valueOf(drsJsonMap.get(DRS_UPDATE_TIMESTAMP));
+                String tableName = StringUtils.remove(String.valueOf(drsJsonMap.get(DRS_TABLE_NAME)), '\"');
+                String operation = String.valueOf(drsJsonMap.get(DRS_OPERATION));
+                String columnCount = String.valueOf(drsJsonMap.get(DRS_COLUMN_COUNT));
+                List<Object> fields = drsJsonMap.get(DRS_FIELDS) != null ? (List<Object>) drsJsonMap.get(DRS_FIELDS) : null;
+                List<Object> beforeImages = drsJsonMap.get(DRS_BEFORE_IMAGES) != null ? (List<Object>) drsJsonMap.get(DRS_BEFORE_IMAGES) : null;
+                List<Object> afterImages = drsJsonMap.get(DRS_AFTER_IMAGES) != null ? (List<Object>) drsJsonMap.get(DRS_AFTER_IMAGES) : null;
+
+                if (fields == null) {
+                    return null;
+                }
+
+                String op = null;
+                if (operation.equalsIgnoreCase("INSERT")) {
+                    op = "c";
+                }
+                if (operation.equalsIgnoreCase("UPDATE")) {
+                    op = "u";
+                }
+                if (operation.equalsIgnoreCase("DELETE")) {
+                    op = "d";
+                }
+                if (op == null) {
+                    return null;
+                }
+
+                Map<String, Object> debeziumJsonMap = new HashMap<>();
+                debeziumJsonMap.put(DEBEZIUM_OP, op);
+                debeziumJsonMap.put(DEBEZIUM_TS_MS, Long.parseLong(updateTimestamp));
+
+                Map<String, Object> debeziumSourceJsonMap = new HashMap<>();
+                debeziumSourceJsonMap.put(DRS_VERSION, version);
+                debeziumSourceJsonMap.put(DRS_SEQ_NO, seqno);
+                debeziumSourceJsonMap.put(DRS_SHARD_ID, shardId);
+                debeziumSourceJsonMap.put(DRS_EVENT_ID, eventId);
+                debeziumSourceJsonMap.put(DRS_UPDATE_TIMESTAMP, updateTimestamp);
+                debeziumSourceJsonMap.put(DRS_OPERATION, op);
+                debeziumSourceJsonMap.put(DRS_TABLE_NAME, tableName);
+                debeziumSourceJsonMap.put(DRS_COLUMN_COUNT, columnCount);
+                debeziumJsonMap.put(DEBEZIUM_SOURCE, debeziumSourceJsonMap);
+
+                List<String> fieldStrList = fields.stream()
+                        .map(field -> ((Map<String, String>) field).get("name")).collect(Collectors.toList());
+
+                if (beforeImages == null) {
+                    debeziumJsonMap.put(DEBEZIUM_BEFORE, null);
+                } else {
+                    List<String> beforeStrList = beforeImages.stream()
+                            .map(before -> {
+                                if (before == null) {
+                                    return null;
+                                }
+                                return ((Map<String, String>) before).get("value");
+                            }).collect(Collectors.toList());
+                    Map<String, String> debeziumBeforeJsonMap = new HashMap<>();
+                    for (int i = 0; i < Integer.parseInt(columnCount); i++) {
+                        debeziumBeforeJsonMap.put(fieldStrList.get(i), beforeStrList.get(i));
+                    }
+                    debeziumJsonMap.put(DEBEZIUM_BEFORE, debeziumBeforeJsonMap);
+                }
+
+                if (afterImages == null) {
+                    debeziumJsonMap.put(DEBEZIUM_AFTER, null);
+                } else {
+                    List<String> afterStrList = afterImages.stream()
+                            .map(after -> {
+                                if (after == null) {
+                                    return null;
+                                }
+                                return ((Map<String, String>) after).get("value");
+                            }).collect(Collectors.toList());
+                    Map<String, String> debeziumAfterJsonMap = new HashMap<>();
+                    for (int i = 0; i < Integer.parseInt(columnCount); i++) {
+                        debeziumAfterJsonMap.put(fieldStrList.get(i), afterStrList.get(i));
+                    }
+                    debeziumJsonMap.put(DEBEZIUM_AFTER, debeziumAfterJsonMap);
+                }
+
+                return debeziumJsonMap;
+            }
+        }).filter(Objects::nonNull).setParallelism(parallelism)
+            .addSink(new FlinkKafkaProducer<>(jsonTopic, (KafkaSerializationSchema<Map<String, Object>>) (jsonMap, aLong) -> {
+                byte[] key = null;
+                byte[] value = null;
+                try {
+                    key = ((Map<String, String>) jsonMap.get(DEBEZIUM_SOURCE)).get(DRS_TABLE_NAME).getBytes();
+                    value = JSON_MAPPER.writeValueAsBytes(jsonMap);
+                } catch (JsonProcessingException e) {
+                    e.printStackTrace();
+                }
+                return new ProducerRecord<>(jsonTopic, key, value);
+        }, props, FlinkKafkaProducer.Semantic.EXACTLY_ONCE));
+        env.execute();
+
+//        DataStreamSource<Map<String, Object>> messageStream =
+//                env.addSource(
+//                        new FlinkKafkaConsumer<>(
+//                                avroTopic, new CustomizedDeSerializationSchema(), props));
+//        messageStream.filter(map -> !map.isEmpty())
+//                .addSink(new FlinkKafkaProducer<>(jsonTopic, new CustomizedSerializationSchema(), props, FlinkKafkaProducer.Semantic.EXACTLY_ONCE));
+//        env.execute();
+
+
+    }
+}

src/flink-examples/flink-examples-security/flink-parse-avro/src/main/java/com/huawei/bigdata/flink/avro/parser/CustomizedDeSerializationSchema.java

@@ -0,0 +1,47 @@
+package com.huawei.bigdata.flink.avro.parser;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.huawei.hwclouds.drs.avro.Record;
+import org.apache.flink.api.common.typeinfo.TypeHint;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+public class CustomizedDeSerializationSchema implements KafkaDeserializationSchema<Map<String, Object>> {
+
+    @Override
+    public boolean isEndOfStream(Map<String, Object> map) {
+        return false;
+    }
+
+    @Override
+    public Map<String, Object> deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) {
+        ObjectMapper mapper = new ObjectMapper();
+        Map<String, Object> map = new HashMap<>();
+        if (consumerRecord != null && consumerRecord.key() != null && consumerRecord.value() != null) {
+            try {
+                Object rec = Record.getDecoder().decode(consumerRecord.value());
+                Map<String, Object> mapAll = mapper.readValue(rec.toString(), new TypeReference<HashMap<String, Object>>(){});
+                map.put(Avro2Json.TOPIC, consumerRecord.topic());
+                map.put(Avro2Json.DRS_TABLE_NAME, mapAll.get(Avro2Json.DRS_TABLE_NAME));
+                map.put(Avro2Json.DRS_UPDATE_TIMESTAMP, mapAll.get(Avro2Json.DRS_UPDATE_TIMESTAMP));
+                map.put(Avro2Json.DRS_OPERATION, mapAll.get(Avro2Json.DRS_OPERATION));
+                map.put("columnNames", mapAll.get(Avro2Json.DRS_FIELDS));
+                map.put("columnValues", mapAll.get(Avro2Json.DRS_AFTER_IMAGES));
+            } catch (IOException exception) {
+                exception.printStackTrace();
+            }
+        }
+        return map;
+    }
+
+    @Override
+    public TypeInformation<Map<String, Object>> getProducedType() {
+        return TypeInformation.of(new TypeHint<Map<String, Object>>(){});
+    }
+}

src/flink-examples/flink-examples-security/flink-parse-avro/src/main/java/com/huawei/bigdata/flink/avro/parser/CustomizedSerializationSchema.java

@@ -0,0 +1,27 @@
+package com.huawei.bigdata.flink.avro.parser;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+import javax.annotation.Nullable;
+import java.util.Map;
+
+public class CustomizedSerializationSchema implements KafkaSerializationSchema<Map<String, Object>> {
+
+    @Override
+    public ProducerRecord<byte[], byte[]> serialize(Map<String, Object> map, @Nullable Long aLong) {
+        ProducerRecord<byte[], byte[]> producerRecord = null;
+        try {
+            ObjectMapper mapper = new ObjectMapper();
+            String topic = (String) map.get(Avro2Json.TOPIC);
+            String tableName = (String) map.get(Avro2Json.DRS_TABLE_NAME);
+            String jsonStr = mapper.writeValueAsString(map);
+            producerRecord = new ProducerRecord<>(topic, tableName.getBytes(), jsonStr.getBytes());
+        } catch (JsonProcessingException e) {
+            e.printStackTrace();
+        }
+        return producerRecord;
+    }
+}

src/flink-examples/flink-examples-security/flink-parse-avro/src/main/java/com/huawei/hwclouds/drs/avro/BinaryObject.java

@@ -0,0 +1,308 @@
+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.huawei.hwclouds.drs.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class BinaryObject extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = -1488248334912497566L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"BinaryObject\",\"namespace\":\"com.huawei.hwclouds.drs.avro\",\"fields\":[{\"name\":\"type\",\"type\":\"string\"},{\"name\":\"value\",\"type\":\"bytes\"}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<BinaryObject> ENCODER =
+      new BinaryMessageEncoder<BinaryObject>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<BinaryObject> DECODER =
+      new BinaryMessageDecoder<BinaryObject>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<BinaryObject> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<BinaryObject> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<BinaryObject>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this BinaryObject to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a BinaryObject from a ByteBuffer. */
+  public static BinaryObject fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  @Deprecated public java.lang.CharSequence type;
+  @Deprecated public java.nio.ByteBuffer value;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public BinaryObject() {}
+
+  /**
+   * All-args constructor.
+   * @param type The new value for type
+   * @param value The new value for value
+   */
+  public BinaryObject(java.lang.CharSequence type, java.nio.ByteBuffer value) {
+    this.type = type;
+    this.value = value;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return type;
+    case 1: return value;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: type = (java.lang.CharSequence)value$; break;
+    case 1: value = (java.nio.ByteBuffer)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'type' field.
+   * @return The value of the 'type' field.
+   */
+  public java.lang.CharSequence getType() {
+    return type;
+  }
+
+  /**
+   * Sets the value of the 'type' field.
+   * @param value the value to set.
+   */
+  public void setType(java.lang.CharSequence value) {
+    this.type = value;
+  }
+
+  /**
+   * Gets the value of the 'value' field.
+   * @return The value of the 'value' field.
+   */
+  public java.nio.ByteBuffer getValue() {
+    return value;
+  }
+
+  /**
+   * Sets the value of the 'value' field.
+   * @param value the value to set.
+   */
+  public void setValue(java.nio.ByteBuffer value) {
+    this.value = value;
+  }
+
+  /**
+   * Creates a new BinaryObject RecordBuilder.
+   * @return A new BinaryObject RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.BinaryObject.Builder newBuilder() {
+    return new com.huawei.hwclouds.drs.avro.BinaryObject.Builder();
+  }
+
+  /**
+   * Creates a new BinaryObject RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new BinaryObject RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.BinaryObject.Builder newBuilder(com.huawei.hwclouds.drs.avro.BinaryObject.Builder other) {
+    return new com.huawei.hwclouds.drs.avro.BinaryObject.Builder(other);
+  }
+
+  /**
+   * Creates a new BinaryObject RecordBuilder by copying an existing BinaryObject instance.
+   * @param other The existing instance to copy.
+   * @return A new BinaryObject RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.BinaryObject.Builder newBuilder(com.huawei.hwclouds.drs.avro.BinaryObject other) {
+    return new com.huawei.hwclouds.drs.avro.BinaryObject.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for BinaryObject instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<BinaryObject>
+    implements org.apache.avro.data.RecordBuilder<BinaryObject> {
+
+    private java.lang.CharSequence type;
+    private java.nio.ByteBuffer value;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.BinaryObject.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.type)) {
+        this.type = data().deepCopy(fields()[0].schema(), other.type);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing BinaryObject instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.BinaryObject other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.type)) {
+        this.type = data().deepCopy(fields()[0].schema(), other.type);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'type' field.
+      * @return The value.
+      */
+    public java.lang.CharSequence getType() {
+      return type;
+    }
+
+    /**
+      * Sets the value of the 'type' field.
+      * @param value The value of 'type'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.BinaryObject.Builder setType(java.lang.CharSequence value) {
+      validate(fields()[0], value);
+      this.type = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'type' field has been set.
+      * @return True if the 'type' field has been set, false otherwise.
+      */
+    public boolean hasType() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'type' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.BinaryObject.Builder clearType() {
+      type = null;
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'value' field.
+      * @return The value.
+      */
+    public java.nio.ByteBuffer getValue() {
+      return value;
+    }
+
+    /**
+      * Sets the value of the 'value' field.
+      * @param value The value of 'value'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.BinaryObject.Builder setValue(java.nio.ByteBuffer value) {
+      validate(fields()[1], value);
+      this.value = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'value' field has been set.
+      * @return True if the 'value' field has been set, false otherwise.
+      */
+    public boolean hasValue() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'value' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.BinaryObject.Builder clearValue() {
+      value = null;
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public BinaryObject build() {
+      try {
+        BinaryObject record = new BinaryObject();
+        record.type = fieldSetFlags()[0] ? this.type : (java.lang.CharSequence) defaultValue(fields()[0]);
+        record.value = fieldSetFlags()[1] ? this.value : (java.nio.ByteBuffer) defaultValue(fields()[1]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<BinaryObject>
+    WRITER$ = (org.apache.avro.io.DatumWriter<BinaryObject>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<BinaryObject>
+    READER$ = (org.apache.avro.io.DatumReader<BinaryObject>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}

src/flink-examples/flink-examples-security/flink-parse-avro/src/main/java/com/huawei/hwclouds/drs/avro/Bit.java

@@ -0,0 +1,307 @@
+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.huawei.hwclouds.drs.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class Bit extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = 6977607147739631720L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Bit\",\"namespace\":\"com.huawei.hwclouds.drs.avro\",\"fields\":[{\"name\":\"precision\",\"type\":\"int\"},{\"name\":\"value\",\"type\":\"string\"}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<Bit> ENCODER =
+      new BinaryMessageEncoder<Bit>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<Bit> DECODER =
+      new BinaryMessageDecoder<Bit>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<Bit> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<Bit> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<Bit>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this Bit to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a Bit from a ByteBuffer. */
+  public static Bit fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  @Deprecated public int precision;
+  @Deprecated public java.lang.CharSequence value;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public Bit() {}
+
+  /**
+   * All-args constructor.
+   * @param precision The new value for precision
+   * @param value The new value for value
+   */
+  public Bit(java.lang.Integer precision, java.lang.CharSequence value) {
+    this.precision = precision;
+    this.value = value;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return precision;
+    case 1: return value;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: precision = (java.lang.Integer)value$; break;
+    case 1: value = (java.lang.CharSequence)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'precision' field.
+   * @return The value of the 'precision' field.
+   */
+  public java.lang.Integer getPrecision() {
+    return precision;
+  }
+
+  /**
+   * Sets the value of the 'precision' field.
+   * @param value the value to set.
+   */
+  public void setPrecision(java.lang.Integer value) {
+    this.precision = value;
+  }
+
+  /**
+   * Gets the value of the 'value' field.
+   * @return The value of the 'value' field.
+   */
+  public java.lang.CharSequence getValue() {
+    return value;
+  }
+
+  /**
+   * Sets the value of the 'value' field.
+   * @param value the value to set.
+   */
+  public void setValue(java.lang.CharSequence value) {
+    this.value = value;
+  }
+
+  /**
+   * Creates a new Bit RecordBuilder.
+   * @return A new Bit RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Bit.Builder newBuilder() {
+    return new com.huawei.hwclouds.drs.avro.Bit.Builder();
+  }
+
+  /**
+   * Creates a new Bit RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new Bit RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Bit.Builder newBuilder(com.huawei.hwclouds.drs.avro.Bit.Builder other) {
+    return new com.huawei.hwclouds.drs.avro.Bit.Builder(other);
+  }
+
+  /**
+   * Creates a new Bit RecordBuilder by copying an existing Bit instance.
+   * @param other The existing instance to copy.
+   * @return A new Bit RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Bit.Builder newBuilder(com.huawei.hwclouds.drs.avro.Bit other) {
+    return new com.huawei.hwclouds.drs.avro.Bit.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for Bit instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<Bit>
+    implements org.apache.avro.data.RecordBuilder<Bit> {
+
+    private int precision;
+    private java.lang.CharSequence value;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Bit.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.precision)) {
+        this.precision = data().deepCopy(fields()[0].schema(), other.precision);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing Bit instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Bit other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.precision)) {
+        this.precision = data().deepCopy(fields()[0].schema(), other.precision);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'precision' field.
+      * @return The value.
+      */
+    public java.lang.Integer getPrecision() {
+      return precision;
+    }
+
+    /**
+      * Sets the value of the 'precision' field.
+      * @param value The value of 'precision'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Bit.Builder setPrecision(int value) {
+      validate(fields()[0], value);
+      this.precision = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'precision' field has been set.
+      * @return True if the 'precision' field has been set, false otherwise.
+      */
+    public boolean hasPrecision() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'precision' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Bit.Builder clearPrecision() {
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'value' field.
+      * @return The value.
+      */
+    public java.lang.CharSequence getValue() {
+      return value;
+    }
+
+    /**
+      * Sets the value of the 'value' field.
+      * @param value The value of 'value'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Bit.Builder setValue(java.lang.CharSequence value) {
+      validate(fields()[1], value);
+      this.value = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'value' field has been set.
+      * @return True if the 'value' field has been set, false otherwise.
+      */
+    public boolean hasValue() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'value' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Bit.Builder clearValue() {
+      value = null;
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public Bit build() {
+      try {
+        Bit record = new Bit();
+        record.precision = fieldSetFlags()[0] ? this.precision : (java.lang.Integer) defaultValue(fields()[0]);
+        record.value = fieldSetFlags()[1] ? this.value : (java.lang.CharSequence) defaultValue(fields()[1]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<Bit>
+    WRITER$ = (org.apache.avro.io.DatumWriter<Bit>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<Bit>
+    READER$ = (org.apache.avro.io.DatumReader<Bit>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}

src/flink-examples/flink-examples-security/flink-parse-avro/src/main/java/com/huawei/hwclouds/drs/avro/Character.java

@@ -0,0 +1,308 @@
+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.huawei.hwclouds.drs.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class Character extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = 8228315679191500556L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Character\",\"namespace\":\"com.huawei.hwclouds.drs.avro\",\"fields\":[{\"name\":\"charset\",\"type\":\"string\"},{\"name\":\"value\",\"type\":\"bytes\"}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<Character> ENCODER =
+      new BinaryMessageEncoder<Character>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<Character> DECODER =
+      new BinaryMessageDecoder<Character>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<Character> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<Character> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<Character>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this Character to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a Character from a ByteBuffer. */
+  public static Character fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  @Deprecated public java.lang.CharSequence charset;
+  @Deprecated public java.nio.ByteBuffer value;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public Character() {}
+
+  /**
+   * All-args constructor.
+   * @param charset The new value for charset
+   * @param value The new value for value
+   */
+  public Character(java.lang.CharSequence charset, java.nio.ByteBuffer value) {
+    this.charset = charset;
+    this.value = value;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return charset;
+    case 1: return value;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: charset = (java.lang.CharSequence)value$; break;
+    case 1: value = (java.nio.ByteBuffer)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'charset' field.
+   * @return The value of the 'charset' field.
+   */
+  public java.lang.CharSequence getCharset() {
+    return charset;
+  }
+
+  /**
+   * Sets the value of the 'charset' field.
+   * @param value the value to set.
+   */
+  public void setCharset(java.lang.CharSequence value) {
+    this.charset = value;
+  }
+
+  /**
+   * Gets the value of the 'value' field.
+   * @return The value of the 'value' field.
+   */
+  public java.nio.ByteBuffer getValue() {
+    return value;
+  }
+
+  /**
+   * Sets the value of the 'value' field.
+   * @param value the value to set.
+   */
+  public void setValue(java.nio.ByteBuffer value) {
+    this.value = value;
+  }
+
+  /**
+   * Creates a new Character RecordBuilder.
+   * @return A new Character RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Character.Builder newBuilder() {
+    return new com.huawei.hwclouds.drs.avro.Character.Builder();
+  }
+
+  /**
+   * Creates a new Character RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new Character RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Character.Builder newBuilder(com.huawei.hwclouds.drs.avro.Character.Builder other) {
+    return new com.huawei.hwclouds.drs.avro.Character.Builder(other);
+  }
+
+  /**
+   * Creates a new Character RecordBuilder by copying an existing Character instance.
+   * @param other The existing instance to copy.
+   * @return A new Character RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Character.Builder newBuilder(com.huawei.hwclouds.drs.avro.Character other) {
+    return new com.huawei.hwclouds.drs.avro.Character.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for Character instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<Character>
+    implements org.apache.avro.data.RecordBuilder<Character> {
+
+    private java.lang.CharSequence charset;
+    private java.nio.ByteBuffer value;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Character.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.charset)) {
+        this.charset = data().deepCopy(fields()[0].schema(), other.charset);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing Character instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Character other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.charset)) {
+        this.charset = data().deepCopy(fields()[0].schema(), other.charset);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'charset' field.
+      * @return The value.
+      */
+    public java.lang.CharSequence getCharset() {
+      return charset;
+    }
+
+    /**
+      * Sets the value of the 'charset' field.
+      * @param value The value of 'charset'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Character.Builder setCharset(java.lang.CharSequence value) {
+      validate(fields()[0], value);
+      this.charset = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'charset' field has been set.
+      * @return True if the 'charset' field has been set, false otherwise.
+      */
+    public boolean hasCharset() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'charset' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Character.Builder clearCharset() {
+      charset = null;
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'value' field.
+      * @return The value.
+      */
+    public java.nio.ByteBuffer getValue() {
+      return value;
+    }
+
+    /**
+      * Sets the value of the 'value' field.
+      * @param value The value of 'value'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Character.Builder setValue(java.nio.ByteBuffer value) {
+      validate(fields()[1], value);
+      this.value = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'value' field has been set.
+      * @return True if the 'value' field has been set, false otherwise.
+      */
+    public boolean hasValue() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'value' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Character.Builder clearValue() {
+      value = null;
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public Character build() {
+      try {
+        Character record = new Character();
+        record.charset = fieldSetFlags()[0] ? this.charset : (java.lang.CharSequence) defaultValue(fields()[0]);
+        record.value = fieldSetFlags()[1] ? this.value : (java.nio.ByteBuffer) defaultValue(fields()[1]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<Character>
+    WRITER$ = (org.apache.avro.io.DatumWriter<Character>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<Character>
+    READER$ = (org.apache.avro.io.DatumReader<Character>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}

src/flink-examples/flink-examples-security/flink-parse-avro/src/main/java/com/huawei/hwclouds/drs/avro/Date.java

@@ -0,0 +1,245 @@
+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.huawei.hwclouds.drs.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class Date extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = 5099846483595947937L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Date\",\"namespace\":\"com.huawei.hwclouds.drs.avro\",\"fields\":[{\"name\":\"value\",\"type\":\"string\",\"doc\":\"Field.dataTypeNumber = 94(DateTime) format:YYYY-MM-DD HH:MM:SS\"}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<Date> ENCODER =
+      new BinaryMessageEncoder<Date>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<Date> DECODER =
+      new BinaryMessageDecoder<Date>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<Date> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<Date> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<Date>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this Date to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a Date from a ByteBuffer. */
+  public static Date fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  /** Field.dataTypeNumber = 94(DateTime) format:YYYY-MM-DD HH:MM:SS */
+  @Deprecated public java.lang.CharSequence value;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public Date() {}
+
+  /**
+   * All-args constructor.
+   * @param value Field.dataTypeNumber = 94(DateTime) format:YYYY-MM-DD HH:MM:SS
+   */
+  public Date(java.lang.CharSequence value) {
+    this.value = value;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return value;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: value = (java.lang.CharSequence)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'value' field.
+   * @return Field.dataTypeNumber = 94(DateTime) format:YYYY-MM-DD HH:MM:SS
+   */
+  public java.lang.CharSequence getValue() {
+    return value;
+  }
+
+  /**
+   * Sets the value of the 'value' field.
+   * Field.dataTypeNumber = 94(DateTime) format:YYYY-MM-DD HH:MM:SS
+   * @param value the value to set.
+   */
+  public void setValue(java.lang.CharSequence value) {
+    this.value = value;
+  }
+
+  /**
+   * Creates a new Date RecordBuilder.
+   * @return A new Date RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Date.Builder newBuilder() {
+    return new com.huawei.hwclouds.drs.avro.Date.Builder();
+  }
+
+  /**
+   * Creates a new Date RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new Date RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Date.Builder newBuilder(com.huawei.hwclouds.drs.avro.Date.Builder other) {
+    return new com.huawei.hwclouds.drs.avro.Date.Builder(other);
+  }
+
+  /**
+   * Creates a new Date RecordBuilder by copying an existing Date instance.
+   * @param other The existing instance to copy.
+   * @return A new Date RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Date.Builder newBuilder(com.huawei.hwclouds.drs.avro.Date other) {
+    return new com.huawei.hwclouds.drs.avro.Date.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for Date instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<Date>
+    implements org.apache.avro.data.RecordBuilder<Date> {
+
+    /** Field.dataTypeNumber = 94(DateTime) format:YYYY-MM-DD HH:MM:SS */
+    private java.lang.CharSequence value;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Date.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.value)) {
+        this.value = data().deepCopy(fields()[0].schema(), other.value);
+        fieldSetFlags()[0] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing Date instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Date other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.value)) {
+        this.value = data().deepCopy(fields()[0].schema(), other.value);
+        fieldSetFlags()[0] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'value' field.
+      * Field.dataTypeNumber = 94(DateTime) format:YYYY-MM-DD HH:MM:SS
+      * @return The value.
+      */
+    public java.lang.CharSequence getValue() {
+      return value;
+    }
+
+    /**
+      * Sets the value of the 'value' field.
+      * Field.dataTypeNumber = 94(DateTime) format:YYYY-MM-DD HH:MM:SS
+      * @param value The value of 'value'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Date.Builder setValue(java.lang.CharSequence value) {
+      validate(fields()[0], value);
+      this.value = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'value' field has been set.
+      * Field.dataTypeNumber = 94(DateTime) format:YYYY-MM-DD HH:MM:SS
+      * @return True if the 'value' field has been set, false otherwise.
+      */
+    public boolean hasValue() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'value' field.
+      * Field.dataTypeNumber = 94(DateTime) format:YYYY-MM-DD HH:MM:SS
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Date.Builder clearValue() {
+      value = null;
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public Date build() {
+      try {
+        Date record = new Date();
+        record.value = fieldSetFlags()[0] ? this.value : (java.lang.CharSequence) defaultValue(fields()[0]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<Date>
+    WRITER$ = (org.apache.avro.io.DatumWriter<Date>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<Date>
+    READER$ = (org.apache.avro.io.DatumReader<Date>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}

src/flink-examples/flink-examples-security/flink-parse-avro/src/main/java/com/huawei/hwclouds/drs/avro/Decimal.java

@@ -0,0 +1,376 @@
+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.huawei.hwclouds.drs.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class Decimal extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = -2192635999292750318L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Decimal\",\"namespace\":\"com.huawei.hwclouds.drs.avro\",\"fields\":[{\"name\":\"value\",\"type\":\"string\"},{\"name\":\"precision\",\"type\":\"int\"},{\"name\":\"scale\",\"type\":\"int\"}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<Decimal> ENCODER =
+      new BinaryMessageEncoder<Decimal>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<Decimal> DECODER =
+      new BinaryMessageDecoder<Decimal>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<Decimal> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<Decimal> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<Decimal>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this Decimal to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a Decimal from a ByteBuffer. */
+  public static Decimal fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  @Deprecated public java.lang.CharSequence value;
+  @Deprecated public int precision;
+  @Deprecated public int scale;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public Decimal() {}
+
+  /**
+   * All-args constructor.
+   * @param value The new value for value
+   * @param precision The new value for precision
+   * @param scale The new value for scale
+   */
+  public Decimal(java.lang.CharSequence value, java.lang.Integer precision, java.lang.Integer scale) {
+    this.value = value;
+    this.precision = precision;
+    this.scale = scale;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return value;
+    case 1: return precision;
+    case 2: return scale;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: value = (java.lang.CharSequence)value$; break;
+    case 1: precision = (java.lang.Integer)value$; break;
+    case 2: scale = (java.lang.Integer)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'value' field.
+   * @return The value of the 'value' field.
+   */
+  public java.lang.CharSequence getValue() {
+    return value;
+  }
+
+  /**
+   * Sets the value of the 'value' field.
+   * @param value the value to set.
+   */
+  public void setValue(java.lang.CharSequence value) {
+    this.value = value;
+  }
+
+  /**
+   * Gets the value of the 'precision' field.
+   * @return The value of the 'precision' field.
+   */
+  public java.lang.Integer getPrecision() {
+    return precision;
+  }
+
+  /**
+   * Sets the value of the 'precision' field.
+   * @param value the value to set.
+   */
+  public void setPrecision(java.lang.Integer value) {
+    this.precision = value;
+  }
+
+  /**
+   * Gets the value of the 'scale' field.
+   * @return The value of the 'scale' field.
+   */
+  public java.lang.Integer getScale() {
+    return scale;
+  }
+
+  /**
+   * Sets the value of the 'scale' field.
+   * @param value the value to set.
+   */
+  public void setScale(java.lang.Integer value) {
+    this.scale = value;
+  }
+
+  /**
+   * Creates a new Decimal RecordBuilder.
+   * @return A new Decimal RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Decimal.Builder newBuilder() {
+    return new com.huawei.hwclouds.drs.avro.Decimal.Builder();
+  }
+
+  /**
+   * Creates a new Decimal RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new Decimal RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Decimal.Builder newBuilder(com.huawei.hwclouds.drs.avro.Decimal.Builder other) {
+    return new com.huawei.hwclouds.drs.avro.Decimal.Builder(other);
+  }
+
+  /**
+   * Creates a new Decimal RecordBuilder by copying an existing Decimal instance.
+   * @param other The existing instance to copy.
+   * @return A new Decimal RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Decimal.Builder newBuilder(com.huawei.hwclouds.drs.avro.Decimal other) {
+    return new com.huawei.hwclouds.drs.avro.Decimal.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for Decimal instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<Decimal>
+    implements org.apache.avro.data.RecordBuilder<Decimal> {
+
+    private java.lang.CharSequence value;
+    private int precision;
+    private int scale;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Decimal.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.value)) {
+        this.value = data().deepCopy(fields()[0].schema(), other.value);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.precision)) {
+        this.precision = data().deepCopy(fields()[1].schema(), other.precision);
+        fieldSetFlags()[1] = true;
+      }
+      if (isValidValue(fields()[2], other.scale)) {
+        this.scale = data().deepCopy(fields()[2].schema(), other.scale);
+        fieldSetFlags()[2] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing Decimal instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Decimal other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.value)) {
+        this.value = data().deepCopy(fields()[0].schema(), other.value);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.precision)) {
+        this.precision = data().deepCopy(fields()[1].schema(), other.precision);
+        fieldSetFlags()[1] = true;
+      }
+      if (isValidValue(fields()[2], other.scale)) {
+        this.scale = data().deepCopy(fields()[2].schema(), other.scale);
+        fieldSetFlags()[2] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'value' field.
+      * @return The value.
+      */
+    public java.lang.CharSequence getValue() {
+      return value;
+    }
+
+    /**
+      * Sets the value of the 'value' field.
+      * @param value The value of 'value'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Decimal.Builder setValue(java.lang.CharSequence value) {
+      validate(fields()[0], value);
+      this.value = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'value' field has been set.
+      * @return True if the 'value' field has been set, false otherwise.
+      */
+    public boolean hasValue() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'value' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Decimal.Builder clearValue() {
+      value = null;
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'precision' field.
+      * @return The value.
+      */
+    public java.lang.Integer getPrecision() {
+      return precision;
+    }
+
+    /**
+      * Sets the value of the 'precision' field.
+      * @param value The value of 'precision'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Decimal.Builder setPrecision(int value) {
+      validate(fields()[1], value);
+      this.precision = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'precision' field has been set.
+      * @return True if the 'precision' field has been set, false otherwise.
+      */
+    public boolean hasPrecision() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'precision' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Decimal.Builder clearPrecision() {
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'scale' field.
+      * @return The value.
+      */
+    public java.lang.Integer getScale() {
+      return scale;
+    }
+
+    /**
+      * Sets the value of the 'scale' field.
+      * @param value The value of 'scale'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Decimal.Builder setScale(int value) {
+      validate(fields()[2], value);
+      this.scale = value;
+      fieldSetFlags()[2] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'scale' field has been set.
+      * @return True if the 'scale' field has been set, false otherwise.
+      */
+    public boolean hasScale() {
+      return fieldSetFlags()[2];
+    }
+
+
+    /**
+      * Clears the value of the 'scale' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Decimal.Builder clearScale() {
+      fieldSetFlags()[2] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public Decimal build() {
+      try {
+        Decimal record = new Decimal();
+        record.value = fieldSetFlags()[0] ? this.value : (java.lang.CharSequence) defaultValue(fields()[0]);
+        record.precision = fieldSetFlags()[1] ? this.precision : (java.lang.Integer) defaultValue(fields()[1]);
+        record.scale = fieldSetFlags()[2] ? this.scale : (java.lang.Integer) defaultValue(fields()[2]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<Decimal>
+    WRITER$ = (org.apache.avro.io.DatumWriter<Decimal>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<Decimal>
+    READER$ = (org.apache.avro.io.DatumReader<Decimal>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}

src/flink-examples/flink-examples-security/flink-parse-avro/src/main/java/com/huawei/hwclouds/drs/avro/Field.java

@@ -0,0 +1,314 @@
+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.huawei.hwclouds.drs.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class Field extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = -7647907005166951319L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Field\",\"namespace\":\"com.huawei.hwclouds.drs.avro\",\"fields\":[{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"dataTypeNumber\",\"type\":\"int\",\"doc\":\"2004, BinaryObject\"}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<Field> ENCODER =
+      new BinaryMessageEncoder<Field>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<Field> DECODER =
+      new BinaryMessageDecoder<Field>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<Field> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<Field> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<Field>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this Field to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a Field from a ByteBuffer. */
+  public static Field fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  @Deprecated public java.lang.CharSequence name;
+  /** 2004, BinaryObject */
+  @Deprecated public int dataTypeNumber;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public Field() {}
+
+  /**
+   * All-args constructor.
+   * @param name The new value for name
+   * @param dataTypeNumber 2004, BinaryObject
+   */
+  public Field(java.lang.CharSequence name, java.lang.Integer dataTypeNumber) {
+    this.name = name;
+    this.dataTypeNumber = dataTypeNumber;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return name;
+    case 1: return dataTypeNumber;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: name = (java.lang.CharSequence)value$; break;
+    case 1: dataTypeNumber = (java.lang.Integer)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'name' field.
+   * @return The value of the 'name' field.
+   */
+  public java.lang.CharSequence getName() {
+    return name;
+  }
+
+  /**
+   * Sets the value of the 'name' field.
+   * @param value the value to set.
+   */
+  public void setName(java.lang.CharSequence value) {
+    this.name = value;
+  }
+
+  /**
+   * Gets the value of the 'dataTypeNumber' field.
+   * @return 2004, BinaryObject
+   */
+  public java.lang.Integer getDataTypeNumber() {
+    return dataTypeNumber;
+  }
+
+  /**
+   * Sets the value of the 'dataTypeNumber' field.
+   * 2004, BinaryObject
+   * @param value the value to set.
+   */
+  public void setDataTypeNumber(java.lang.Integer value) {
+    this.dataTypeNumber = value;
+  }
+
+  /**
+   * Creates a new Field RecordBuilder.
+   * @return A new Field RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Field.Builder newBuilder() {
+    return new com.huawei.hwclouds.drs.avro.Field.Builder();
+  }
+
+  /**
+   * Creates a new Field RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new Field RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Field.Builder newBuilder(com.huawei.hwclouds.drs.avro.Field.Builder other) {
+    return new com.huawei.hwclouds.drs.avro.Field.Builder(other);
+  }
+
+  /**
+   * Creates a new Field RecordBuilder by copying an existing Field instance.
+   * @param other The existing instance to copy.
+   * @return A new Field RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Field.Builder newBuilder(com.huawei.hwclouds.drs.avro.Field other) {
+    return new com.huawei.hwclouds.drs.avro.Field.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for Field instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<Field>
+    implements org.apache.avro.data.RecordBuilder<Field> {
+
+    private java.lang.CharSequence name;
+    /** 2004, BinaryObject */
+    private int dataTypeNumber;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Field.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.name)) {
+        this.name = data().deepCopy(fields()[0].schema(), other.name);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.dataTypeNumber)) {
+        this.dataTypeNumber = data().deepCopy(fields()[1].schema(), other.dataTypeNumber);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing Field instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Field other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.name)) {
+        this.name = data().deepCopy(fields()[0].schema(), other.name);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.dataTypeNumber)) {
+        this.dataTypeNumber = data().deepCopy(fields()[1].schema(), other.dataTypeNumber);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'name' field.
+      * @return The value.
+      */
+    public java.lang.CharSequence getName() {
+      return name;
+    }
+
+    /**
+      * Sets the value of the 'name' field.
+      * @param value The value of 'name'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Field.Builder setName(java.lang.CharSequence value) {
+      validate(fields()[0], value);
+      this.name = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'name' field has been set.
+      * @return True if the 'name' field has been set, false otherwise.
+      */
+    public boolean hasName() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'name' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Field.Builder clearName() {
+      name = null;
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'dataTypeNumber' field.
+      * 2004, BinaryObject
+      * @return The value.
+      */
+    public java.lang.Integer getDataTypeNumber() {
+      return dataTypeNumber;
+    }
+
+    /**
+      * Sets the value of the 'dataTypeNumber' field.
+      * 2004, BinaryObject
+      * @param value The value of 'dataTypeNumber'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Field.Builder setDataTypeNumber(int value) {
+      validate(fields()[1], value);
+      this.dataTypeNumber = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'dataTypeNumber' field has been set.
+      * 2004, BinaryObject
+      * @return True if the 'dataTypeNumber' field has been set, false otherwise.
+      */
+    public boolean hasDataTypeNumber() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'dataTypeNumber' field.
+      * 2004, BinaryObject
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Field.Builder clearDataTypeNumber() {
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public Field build() {
+      try {
+        Field record = new Field();
+        record.name = fieldSetFlags()[0] ? this.name : (java.lang.CharSequence) defaultValue(fields()[0]);
+        record.dataTypeNumber = fieldSetFlags()[1] ? this.dataTypeNumber : (java.lang.Integer) defaultValue(fields()[1]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<Field>
+    WRITER$ = (org.apache.avro.io.DatumWriter<Field>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<Field>
+    READER$ = (org.apache.avro.io.DatumReader<Field>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}

src/flink-examples/flink-examples-security/flink-parse-avro/src/main/java/com/huawei/hwclouds/drs/avro/Float.java

@@ -0,0 +1,375 @@
+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.huawei.hwclouds.drs.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class Float extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = -345095849001738989L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Float\",\"namespace\":\"com.huawei.hwclouds.drs.avro\",\"fields\":[{\"name\":\"value\",\"type\":\"double\"},{\"name\":\"precision\",\"type\":\"int\"},{\"name\":\"scale\",\"type\":\"int\"}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<Float> ENCODER =
+      new BinaryMessageEncoder<Float>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<Float> DECODER =
+      new BinaryMessageDecoder<Float>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<Float> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<Float> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<Float>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this Float to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a Float from a ByteBuffer. */
+  public static Float fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  @Deprecated public double value;
+  @Deprecated public int precision;
+  @Deprecated public int scale;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public Float() {}
+
+  /**
+   * All-args constructor.
+   * @param value The new value for value
+   * @param precision The new value for precision
+   * @param scale The new value for scale
+   */
+  public Float(java.lang.Double value, java.lang.Integer precision, java.lang.Integer scale) {
+    this.value = value;
+    this.precision = precision;
+    this.scale = scale;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return value;
+    case 1: return precision;
+    case 2: return scale;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: value = (java.lang.Double)value$; break;
+    case 1: precision = (java.lang.Integer)value$; break;
+    case 2: scale = (java.lang.Integer)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'value' field.
+   * @return The value of the 'value' field.
+   */
+  public java.lang.Double getValue() {
+    return value;
+  }
+
+  /**
+   * Sets the value of the 'value' field.
+   * @param value the value to set.
+   */
+  public void setValue(java.lang.Double value) {
+    this.value = value;
+  }
+
+  /**
+   * Gets the value of the 'precision' field.
+   * @return The value of the 'precision' field.
+   */
+  public java.lang.Integer getPrecision() {
+    return precision;
+  }
+
+  /**
+   * Sets the value of the 'precision' field.
+   * @param value the value to set.
+   */
+  public void setPrecision(java.lang.Integer value) {
+    this.precision = value;
+  }
+
+  /**
+   * Gets the value of the 'scale' field.
+   * @return The value of the 'scale' field.
+   */
+  public java.lang.Integer getScale() {
+    return scale;
+  }
+
+  /**
+   * Sets the value of the 'scale' field.
+   * @param value the value to set.
+   */
+  public void setScale(java.lang.Integer value) {
+    this.scale = value;
+  }
+
+  /**
+   * Creates a new Float RecordBuilder.
+   * @return A new Float RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Float.Builder newBuilder() {
+    return new com.huawei.hwclouds.drs.avro.Float.Builder();
+  }
+
+  /**
+   * Creates a new Float RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new Float RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Float.Builder newBuilder(com.huawei.hwclouds.drs.avro.Float.Builder other) {
+    return new com.huawei.hwclouds.drs.avro.Float.Builder(other);
+  }
+
+  /**
+   * Creates a new Float RecordBuilder by copying an existing Float instance.
+   * @param other The existing instance to copy.
+   * @return A new Float RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Float.Builder newBuilder(com.huawei.hwclouds.drs.avro.Float other) {
+    return new com.huawei.hwclouds.drs.avro.Float.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for Float instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<Float>
+    implements org.apache.avro.data.RecordBuilder<Float> {
+
+    private double value;
+    private int precision;
+    private int scale;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Float.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.value)) {
+        this.value = data().deepCopy(fields()[0].schema(), other.value);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.precision)) {
+        this.precision = data().deepCopy(fields()[1].schema(), other.precision);
+        fieldSetFlags()[1] = true;
+      }
+      if (isValidValue(fields()[2], other.scale)) {
+        this.scale = data().deepCopy(fields()[2].schema(), other.scale);
+        fieldSetFlags()[2] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing Float instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Float other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.value)) {
+        this.value = data().deepCopy(fields()[0].schema(), other.value);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.precision)) {
+        this.precision = data().deepCopy(fields()[1].schema(), other.precision);
+        fieldSetFlags()[1] = true;
+      }
+      if (isValidValue(fields()[2], other.scale)) {
+        this.scale = data().deepCopy(fields()[2].schema(), other.scale);
+        fieldSetFlags()[2] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'value' field.
+      * @return The value.
+      */
+    public java.lang.Double getValue() {
+      return value;
+    }
+
+    /**
+      * Sets the value of the 'value' field.
+      * @param value The value of 'value'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Float.Builder setValue(double value) {
+      validate(fields()[0], value);
+      this.value = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'value' field has been set.
+      * @return True if the 'value' field has been set, false otherwise.
+      */
+    public boolean hasValue() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'value' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Float.Builder clearValue() {
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'precision' field.
+      * @return The value.
+      */
+    public java.lang.Integer getPrecision() {
+      return precision;
+    }
+
+    /**
+      * Sets the value of the 'precision' field.
+      * @param value The value of 'precision'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Float.Builder setPrecision(int value) {
+      validate(fields()[1], value);
+      this.precision = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'precision' field has been set.
+      * @return True if the 'precision' field has been set, false otherwise.
+      */
+    public boolean hasPrecision() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'precision' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Float.Builder clearPrecision() {
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'scale' field.
+      * @return The value.
+      */
+    public java.lang.Integer getScale() {
+      return scale;
+    }
+
+    /**
+      * Sets the value of the 'scale' field.
+      * @param value The value of 'scale'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Float.Builder setScale(int value) {
+      validate(fields()[2], value);
+      this.scale = value;
+      fieldSetFlags()[2] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'scale' field has been set.
+      * @return True if the 'scale' field has been set, false otherwise.
+      */
+    public boolean hasScale() {
+      return fieldSetFlags()[2];
+    }
+
+
+    /**
+      * Clears the value of the 'scale' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Float.Builder clearScale() {
+      fieldSetFlags()[2] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public Float build() {
+      try {
+        Float record = new Float();
+        record.value = fieldSetFlags()[0] ? this.value : (java.lang.Double) defaultValue(fields()[0]);
+        record.precision = fieldSetFlags()[1] ? this.precision : (java.lang.Integer) defaultValue(fields()[1]);
+        record.scale = fieldSetFlags()[2] ? this.scale : (java.lang.Integer) defaultValue(fields()[2]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<Float>
+    WRITER$ = (org.apache.avro.io.DatumWriter<Float>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<Float>
+    READER$ = (org.apache.avro.io.DatumReader<Float>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}

src/flink-examples/flink-examples-security/flink-parse-avro/src/main/java/com/huawei/hwclouds/drs/avro/Integer.java

@@ -0,0 +1,307 @@
+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.huawei.hwclouds.drs.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class Integer extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = -5960734445972534049L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Integer\",\"namespace\":\"com.huawei.hwclouds.drs.avro\",\"fields\":[{\"name\":\"precision\",\"type\":\"int\"},{\"name\":\"value\",\"type\":\"string\"}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<Integer> ENCODER =
+      new BinaryMessageEncoder<Integer>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<Integer> DECODER =
+      new BinaryMessageDecoder<Integer>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<Integer> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<Integer> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<Integer>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this Integer to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a Integer from a ByteBuffer. */
+  public static Integer fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  @Deprecated public int precision;
+  @Deprecated public java.lang.CharSequence value;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public Integer() {}
+
+  /**
+   * All-args constructor.
+   * @param precision The new value for precision
+   * @param value The new value for value
+   */
+  public Integer(java.lang.Integer precision, java.lang.CharSequence value) {
+    this.precision = precision;
+    this.value = value;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return precision;
+    case 1: return value;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: precision = (java.lang.Integer)value$; break;
+    case 1: value = (java.lang.CharSequence)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'precision' field.
+   * @return The value of the 'precision' field.
+   */
+  public java.lang.Integer getPrecision() {
+    return precision;
+  }
+
+  /**
+   * Sets the value of the 'precision' field.
+   * @param value the value to set.
+   */
+  public void setPrecision(java.lang.Integer value) {
+    this.precision = value;
+  }
+
+  /**
+   * Gets the value of the 'value' field.
+   * @return The value of the 'value' field.
+   */
+  public java.lang.CharSequence getValue() {
+    return value;
+  }
+
+  /**
+   * Sets the value of the 'value' field.
+   * @param value the value to set.
+   */
+  public void setValue(java.lang.CharSequence value) {
+    this.value = value;
+  }
+
+  /**
+   * Creates a new Integer RecordBuilder.
+   * @return A new Integer RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Integer.Builder newBuilder() {
+    return new com.huawei.hwclouds.drs.avro.Integer.Builder();
+  }
+
+  /**
+   * Creates a new Integer RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new Integer RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Integer.Builder newBuilder(com.huawei.hwclouds.drs.avro.Integer.Builder other) {
+    return new com.huawei.hwclouds.drs.avro.Integer.Builder(other);
+  }
+
+  /**
+   * Creates a new Integer RecordBuilder by copying an existing Integer instance.
+   * @param other The existing instance to copy.
+   * @return A new Integer RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Integer.Builder newBuilder(com.huawei.hwclouds.drs.avro.Integer other) {
+    return new com.huawei.hwclouds.drs.avro.Integer.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for Integer instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<Integer>
+    implements org.apache.avro.data.RecordBuilder<Integer> {
+
+    private int precision;
+    private java.lang.CharSequence value;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Integer.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.precision)) {
+        this.precision = data().deepCopy(fields()[0].schema(), other.precision);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing Integer instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Integer other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.precision)) {
+        this.precision = data().deepCopy(fields()[0].schema(), other.precision);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'precision' field.
+      * @return The value.
+      */
+    public java.lang.Integer getPrecision() {
+      return precision;
+    }
+
+    /**
+      * Sets the value of the 'precision' field.
+      * @param value The value of 'precision'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Integer.Builder setPrecision(int value) {
+      validate(fields()[0], value);
+      this.precision = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'precision' field has been set.
+      * @return True if the 'precision' field has been set, false otherwise.
+      */
+    public boolean hasPrecision() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'precision' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Integer.Builder clearPrecision() {
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'value' field.
+      * @return The value.
+      */
+    public java.lang.CharSequence getValue() {
+      return value;
+    }
+
+    /**
+      * Sets the value of the 'value' field.
+      * @param value The value of 'value'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Integer.Builder setValue(java.lang.CharSequence value) {
+      validate(fields()[1], value);
+      this.value = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'value' field has been set.
+      * @return True if the 'value' field has been set, false otherwise.
+      */
+    public boolean hasValue() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'value' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Integer.Builder clearValue() {
+      value = null;
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public Integer build() {
+      try {
+        Integer record = new Integer();
+        record.precision = fieldSetFlags()[0] ? this.precision : (java.lang.Integer) defaultValue(fields()[0]);
+        record.value = fieldSetFlags()[1] ? this.value : (java.lang.CharSequence) defaultValue(fields()[1]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<Integer>
+    WRITER$ = (org.apache.avro.io.DatumWriter<Integer>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<Integer>
+    READER$ = (org.apache.avro.io.DatumReader<Integer>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}

src/flink-examples/flink-examples-security/flink-parse-avro/src/main/java/com/huawei/hwclouds/drs/avro/Timestamp.java

@@ -0,0 +1,306 @@
+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.huawei.hwclouds.drs.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class Timestamp extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = 4308662166148827007L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Timestamp\",\"namespace\":\"com.huawei.hwclouds.drs.avro\",\"fields\":[{\"name\":\"timestamp\",\"type\":\"long\"},{\"name\":\"millis\",\"type\":\"int\"}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<Timestamp> ENCODER =
+      new BinaryMessageEncoder<Timestamp>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<Timestamp> DECODER =
+      new BinaryMessageDecoder<Timestamp>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<Timestamp> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<Timestamp> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<Timestamp>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this Timestamp to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a Timestamp from a ByteBuffer. */
+  public static Timestamp fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  @Deprecated public long timestamp;
+  @Deprecated public int millis;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public Timestamp() {}
+
+  /**
+   * All-args constructor.
+   * @param timestamp The new value for timestamp
+   * @param millis The new value for millis
+   */
+  public Timestamp(java.lang.Long timestamp, java.lang.Integer millis) {
+    this.timestamp = timestamp;
+    this.millis = millis;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return timestamp;
+    case 1: return millis;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: timestamp = (java.lang.Long)value$; break;
+    case 1: millis = (java.lang.Integer)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'timestamp' field.
+   * @return The value of the 'timestamp' field.
+   */
+  public java.lang.Long getTimestamp() {
+    return timestamp;
+  }
+
+  /**
+   * Sets the value of the 'timestamp' field.
+   * @param value the value to set.
+   */
+  public void setTimestamp(java.lang.Long value) {
+    this.timestamp = value;
+  }
+
+  /**
+   * Gets the value of the 'millis' field.
+   * @return The value of the 'millis' field.
+   */
+  public java.lang.Integer getMillis() {
+    return millis;
+  }
+
+  /**
+   * Sets the value of the 'millis' field.
+   * @param value the value to set.
+   */
+  public void setMillis(java.lang.Integer value) {
+    this.millis = value;
+  }
+
+  /**
+   * Creates a new Timestamp RecordBuilder.
+   * @return A new Timestamp RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Timestamp.Builder newBuilder() {
+    return new com.huawei.hwclouds.drs.avro.Timestamp.Builder();
+  }
+
+  /**
+   * Creates a new Timestamp RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new Timestamp RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Timestamp.Builder newBuilder(com.huawei.hwclouds.drs.avro.Timestamp.Builder other) {
+    return new com.huawei.hwclouds.drs.avro.Timestamp.Builder(other);
+  }
+
+  /**
+   * Creates a new Timestamp RecordBuilder by copying an existing Timestamp instance.
+   * @param other The existing instance to copy.
+   * @return A new Timestamp RecordBuilder
+   */
+  public static com.huawei.hwclouds.drs.avro.Timestamp.Builder newBuilder(com.huawei.hwclouds.drs.avro.Timestamp other) {
+    return new com.huawei.hwclouds.drs.avro.Timestamp.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for Timestamp instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<Timestamp>
+    implements org.apache.avro.data.RecordBuilder<Timestamp> {
+
+    private long timestamp;
+    private int millis;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Timestamp.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.timestamp)) {
+        this.timestamp = data().deepCopy(fields()[0].schema(), other.timestamp);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.millis)) {
+        this.millis = data().deepCopy(fields()[1].schema(), other.millis);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing Timestamp instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.huawei.hwclouds.drs.avro.Timestamp other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.timestamp)) {
+        this.timestamp = data().deepCopy(fields()[0].schema(), other.timestamp);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.millis)) {
+        this.millis = data().deepCopy(fields()[1].schema(), other.millis);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'timestamp' field.
+      * @return The value.
+      */
+    public java.lang.Long getTimestamp() {
+      return timestamp;
+    }
+
+    /**
+      * Sets the value of the 'timestamp' field.
+      * @param value The value of 'timestamp'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Timestamp.Builder setTimestamp(long value) {
+      validate(fields()[0], value);
+      this.timestamp = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'timestamp' field has been set.
+      * @return True if the 'timestamp' field has been set, false otherwise.
+      */
+    public boolean hasTimestamp() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'timestamp' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Timestamp.Builder clearTimestamp() {
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'millis' field.
+      * @return The value.
+      */
+    public java.lang.Integer getMillis() {
+      return millis;
+    }
+
+    /**
+      * Sets the value of the 'millis' field.
+      * @param value The value of 'millis'.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Timestamp.Builder setMillis(int value) {
+      validate(fields()[1], value);
+      this.millis = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'millis' field has been set.
+      * @return True if the 'millis' field has been set, false otherwise.
+      */
+    public boolean hasMillis() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'millis' field.
+      * @return This builder.
+      */
+    public com.huawei.hwclouds.drs.avro.Timestamp.Builder clearMillis() {
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public Timestamp build() {
+      try {
+        Timestamp record = new Timestamp();
+        record.timestamp = fieldSetFlags()[0] ? this.timestamp : (java.lang.Long) defaultValue(fields()[0]);
+        record.millis = fieldSetFlags()[1] ? this.millis : (java.lang.Integer) defaultValue(fields()[1]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<Timestamp>
+    WRITER$ = (org.apache.avro.io.DatumWriter<Timestamp>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<Timestamp>
+    READER$ = (org.apache.avro.io.DatumReader<Timestamp>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}