/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.api.common.serialization;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.functions.InvalidTypesException;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.util.FlinkRuntimeException;

import java.io.IOException;

import static org.apache.flink.util.Preconditions.checkNotNull;

/**
 * The deserialization schema describes how to turn the byte messages delivered by certain data
 * sources (for example Apache Kafka) into data types (Java/Scala objects) that are processed by
 * Flink.
 *
 * <p>This base variant of the deserialization schema produces the type information automatically by
 * extracting it from the generic class arguments.
 *
 * <h3>Common Use</h3>
 *
 * <p>To write a deserialization schema for a specific type, simply extend this class and declare
 * the type in the class signature. Flink will reflectively determine the type and create the proper
 * TypeInformation:
 *
 * <pre>{@code
 * public class MyDeserializationSchema extends AbstractDeserializationSchema<MyType> {
 *
 *     public MyType deserialize(byte[] message) throws IOException {
 *         ...
 *     }
 * }
 * }</pre>
 *
 * <h3>Generic Use</h3>
 *
 * <p>If you want to write a more generic DeserializationSchema that works for different types, you
 * need to pass the TypeInformation (or an equivalent hint) to the constructor:
 *
 * <pre>{@code
 * public class MyGenericSchema<T> extends AbstractDeserializationSchema<T> {
 *
 *     public MyGenericSchema(Class<T> type) {
 *         super(type);
 *     }
 *
 *     public T deserialize(byte[] message) throws IOException {
 *         ...
 *     }
 * }
 * }</pre>
 *
 * @param <T> The type created by the deserialization schema.
 */
@PublicEvolving
public abstract class AbstractDeserializationSchema<T> implements DeserializationSchema<T> {

    private static final long serialVersionUID = 2L;

    /** The type produced by this {@code DeserializationSchema}. */
    private final TypeInformation<T> type;

    // ------------------------------------------------------------------------

    /**
     * Creates a new AbstractDeserializationSchema and tries to infer the type returned by this
     * DeserializationSchema.
     *
     * <p>This constructor is usable whenever the DeserializationSchema concretely defines its type,
     * without generic variables:
     *
     * <pre>{@code
     * public class MyDeserializationSchema extends AbstractDeserializationSchema<MyType> {
     *
     *     public MyType deserialize(byte[] message) throws IOException {
     *         ...
     *     }
     * }
     * }</pre>
     */
    protected AbstractDeserializationSchema() {
        try {
            this.type =
                    TypeExtractor.createTypeInfo(
                            AbstractDeserializationSchema.class, getClass(), 0, null, null);
        } catch (InvalidTypesException e) {
            throw new FlinkRuntimeException(
                    "The implementation of AbstractDeserializationSchema is using a generic variable. "
                            + "This is not supported, because due to Java's generic type erasure, it will not be possible to "
                            + "determine the full type at runtime. For generic implementations, please pass the TypeInformation "
                            + "or type class explicitly to the constructor.");
        }
    }

    /**
     * Creates an AbstractDeserializationSchema that returns the TypeInformation indicated by the
     * given class. This constructor is only necessary when creating a generic implementation, see
     * {@link AbstractDeserializationSchema Generic Use}.
     *
     * <p>This constructor may fail if the class is generic. In that case, please use the
     * constructor that accepts a {@link #AbstractDeserializationSchema(TypeHint) TypeHint}, or a
     * {@link #AbstractDeserializationSchema(TypeInformation) TypeInformation}.
     *
     * @param type The class of the produced type.
     */
    protected AbstractDeserializationSchema(Class<T> type) {
        checkNotNull(type, "type");
        this.type = TypeInformation.of(type);
    }

    /**
     * Creates an AbstractDeserializationSchema that returns the TypeInformation indicated by the
     * given type hint. This constructor is only necessary when creating a generic implementation,
     * see {@link AbstractDeserializationSchema Generic Use}.
     *
     * @param typeHint The TypeHint for the produced type.
     */
    protected AbstractDeserializationSchema(TypeHint<T> typeHint) {
        checkNotNull(typeHint, "typeHint");
        this.type = typeHint.getTypeInfo();
    }

    /**
     * Creates an AbstractDeserializationSchema that returns the given TypeInformation for the
     * produced type. This constructor is only necessary when creating a generic implementation, see
     * {@link AbstractDeserializationSchema Generic Use}.
     *
     * @param typeInfo The TypeInformation for the produced type.
     */
    protected AbstractDeserializationSchema(TypeInformation<T> typeInfo) {
        this.type = checkNotNull(typeInfo, "typeInfo");
    }

    // ------------------------------------------------------------------------

    /**
     * De-serializes the byte message.
     *
     * @param message The message, as a byte array.
     * @return The de-serialized message as an object.
     */
    @Override
    public abstract T deserialize(byte[] message) throws IOException;

    /**
     * Method to decide whether the element signals the end of the stream. If true is returned the
     * element won't be emitted.
     *
     * <p>This default implementation returns always false, meaning the stream is interpreted to be
     * unbounded.
     *
     * @param nextElement The element to test for the end-of-stream signal.
     * @return True, if the element signals end of stream, false otherwise.
     */
    @Override
    public boolean isEndOfStream(T nextElement) {
        return false;
    }

    /**
     * Gets the type produced by this deserializer. This is the type that was passed to the
     * constructor, or reflectively inferred (if the default constructor was called).
     */
    @Override
    public TypeInformation<T> getProducedType() {
        return type;
    }
}

@@ -4333,16 +4333,39 @@
 ructor.%22
+,%0A                    e
 );%0A     

Add cause when AbstractDeserializationSchema throw exception