Flink 1.14.0 消费 kafka 数据自定义反序列化类

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers(brokers)
    .setTopics("input-topic")
    .setGroupId("my-group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();
env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

/**
 * Sets the {@link KafkaRecordDeserializationSchema deserializer} of the {@link
 * org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} for KafkaSource.
 *
 * @param recordDeserializer the deserializer for Kafka {@link
 *     org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord}.
 * @return this KafkaSourceBuilder.
 */
public KafkaSourceBuilder<OUT> setDeserializer(
        KafkaRecordDeserializationSchema<OUT> recordDeserializer) {
    this.deserializationSchema = recordDeserializer;
    return this;
}
/**
 * Sets the {@link KafkaRecordDeserializationSchema deserializer} of the {@link
 * org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} for KafkaSource. The given
 * {@link DeserializationSchema} will be used to deserialize the value of ConsumerRecord. The
 * other information (e.g. key) in a ConsumerRecord will be ignored.
 *
 * @param deserializationSchema the {@link DeserializationSchema} to use for deserialization.
 * @return this KafkaSourceBuilder.
 */
public KafkaSourceBuilder<OUT> setValueOnlyDeserializer(
        DeserializationSchema<OUT> deserializationSchema) {
    this.deserializationSchema =
            KafkaRecordDeserializationSchema.valueOnly(deserializationSchema);
    return this;
}

@Public
public interface DeserializationSchema<T> extends Serializable, ResultTypeQueryable<T> {
    @PublicEvolving
    default void open(InitializationContext context) throws Exception {}
    T deserialize(byte[] message) throws IOException;
    @PublicEvolving
    default void deserialize(byte[] message, Collector<T> out) throws IOException {
        T deserialize = deserialize(message);
        if (deserialize != null) {
            out.collect(deserialize);
        }
    }
    boolean isEndOfStream(T nextElement);
}

/** An interface for the deserialization of Kafka records. */
public interface KafkaRecordDeserializationSchema<T> extends Serializable, ResultTypeQueryable<T> {
    @PublicEvolving
    default void open(DeserializationSchema.InitializationContext context) throws Exception {}
    @PublicEvolving
    void deserialize(ConsumerRecord<byte[], byte[]> record, Collector<T> out) throws IOException;
    static <V> KafkaRecordDeserializationSchema<V> of(
            KafkaDeserializationSchema<V> kafkaDeserializationSchema) {
        return new KafkaDeserializationSchemaWrapper<>(kafkaDeserializationSchema);
    }
    static <V> KafkaRecordDeserializationSchema<V> valueOnly(
            DeserializationSchema<V> valueDeserializationSchema) {
        return new KafkaValueOnlyDeserializationSchemaWrapper<>(valueDeserializationSchema);
    }
    static <V> KafkaRecordDeserializationSchema<V> valueOnly(
            Class<? extends Deserializer<V>> valueDeserializerClass) {
        return new KafkaValueOnlyDeserializerWrapper<>(
                valueDeserializerClass, Collections.emptyMap());
    }
    static <V, D extends Configurable & Deserializer<V>>
            KafkaRecordDeserializationSchema<V> valueOnly(
                    Class<D> valueDeserializerClass, Map<String, String> config) {
        return new KafkaValueOnlyDeserializerWrapper<>(valueDeserializerClass, config);
    }
}

package flink.stream.deserialization;
import bean.Jason;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.log4j.Logger;
public class MyKafkaDeserialization implements KafkaDeserializationSchema<Jason> {
    private static final Logger log = Logger.getLogger(MyKafkaDeserialization.class);
    private final String encoding = "UTF8";
    private boolean includeTopic;
    private boolean includeTimestamp;
    public MyKafkaDeserialization(boolean includeTopic, boolean includeTimestamp) {
        this.includeTopic = includeTopic;
        this.includeTimestamp = includeTimestamp;
    }
    @Override
    public TypeInformation<Jason> getProducedType() {
        return TypeInformation.of(Jason.class);
    }
    @Override
    public boolean isEndOfStream(Jason nextElement) {
        return false;
    }
    @Override
    public Jason deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
        if (consumerRecord != null) {
            try {
                String value = new String(consumerRecord.value(), encoding);
                Jason jason = JSON.parseObject(value, Jason.class);
                if (includeTopic) jason.setTopic(consumerRecord.topic());
                if (includeTimestamp) jason.setTimestamp(consumerRecord.timestamp());
                return jason;
            } catch (Exception e) {
                log.error("deserialize failed : " + e.getMessage());
            }
        }
        return null;
    }
}

KafkaSource<Jason> source = KafkaSource.<Jason>builder()
        .setProperty("security.protocol", "SASL_PLAINTEXT")
        .setProperty("sasl.mechanism", "PLAIN")
        .setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + username + "\" password=\"" + password + "\";")
        // discover new partitions per 10 seconds
        .setProperty("partition.discovery.interval.ms", "10000")
        .setBootstrapServers(broker)
        .setTopics(topic)
        .setGroupId(group_id)
        .setStartingOffsets(OffsetsInitializer.earliest())
        .setDeserializer(KafkaRecordDeserializationSchema.of(new MyKafkaDeserialization(true, true)))
        // 只反序列化 value
        .setValueOnlyDeserializer(new MyDeSerializer())
        .build();