class FlinkKinesisProducer extends RichSinkFunction<OUT> implements CheckpointedFunction {

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);

		// check and pass the configuration properties
		KinesisProducerConfiguration producerConfig = KinesisConfigUtil.getValidatedProducerConfiguration(configProps);

		producer = getKinesisProducer(producerConfig);
		callback = new FutureCallback<UserRecordResult>() {
			@Override
			public void onSuccess(UserRecordResult result) {
				if (!result.isSuccessful()) {
					if (failOnError) {
						// only remember the first thrown exception
						if (thrownException == null) {
							thrownException = new RuntimeException("Record was not sent successful");
						}
					} else {
						LOG.warn("Record was not sent successful");
					}
				}
			}

			@Override
			public void onFailure(Throwable t) {
				if (failOnError) {
					thrownException = t;
				} else {
					LOG.warn("An exception occurred while processing a record", t);
				}
			}
		};

		if (this.customPartitioner != null) {
			this.customPartitioner.initialize(getRuntimeContext().getIndexOfThisSubtask(), getRuntimeContext().getNumberOfParallelSubtasks());
		}

		LOG.info("Started Kinesis producer instance for region '{}'", producerConfig.getRegion());
	}

	public  FlinkKinesisProducer(final SerializationSchema<OUT> schema, Properties configProps);
	public  FlinkKinesisProducer(KinesisSerializationSchema<OUT> schema, Properties configProps);

}