FlinkKinesisFirehoseProducer does not call SerializationSchema#open

[bobtiernay-okta]

Currently, the implementation of FlinkKinesisFirehoseProducer does not call SerializationSchema#open. This causes problems when a client SerializationSchema needs to perform some initialization on open (e.g. initialize metrics, etc.)

[bobtiernay-okta]

To get around this issue for now, I'm doing the following:

/**
 * See <a href="https://github.com/aws/aws-kinesisanalytics-flink-connectors/issues/14">aws-kinesisanalytics-flink-connector #14</a>
 * for description of why this is needed.
 *
 * @param <T> the type to serialize and produce.
 */
public class CustomFlinkKinesisFirehoseProducer<T> extends FlinkKinesisFirehoseProducer<T> {

    private static final long serialVersionUID = 1L;

    private final SerializationSchema<T> schema;

    public CustomFlinkKinesisFirehoseProducer(String deliveryStream, SerializationSchema<T> schema, Properties configProps) {
        super(deliveryStream, schema, configProps);
        this.schema = schema;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        schema.open(() -> getRuntimeContext().getMetricGroup());
    }

}

[dannycranmer]

Thanks for reporting the issue, we will include in the next release.