org.apache.flink.streaming.connectors.kafka

Class FlinkKafkaProducerBase<IN>

java.lang.Object

org.apache.flink.api.common.functions.AbstractRichFunction

org.apache.flink.streaming.api.functions.sink.RichSinkFunction<IN>

org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase<IN>

Type Parameters:

IN - Type of the messages to write into Kafka.

All Implemented Interfaces:

Serializable, Function, RichFunction, CheckpointedFunction, SinkFunction<IN>

Direct Known Subclasses:

FlinkKafkaProducer010

@Internal
public abstract class FlinkKafkaProducerBase<IN>
extends RichSinkFunction<IN>
implements CheckpointedFunction

Flink Sink to produce data into a Kafka topic.

Please note that this producer provides at-least-once reliability guarantees when checkpoints are enabled and setFlushOnCheckpoint(true) is set. Otherwise, the producer doesn't provide any reliability guarantees.

See Also:

Serialized Form

Nested Class Summary

Nested classes/interfaces inherited from interface org.apache.flink.streaming.api.functions.sink.SinkFunction

SinkFunction.Context<T>

Field Summary

Fields

Modifier and Type	Field and Description
protected Exception	asyncException Errors encountered in the async producer are stored here.
protected org.apache.kafka.clients.producer.Callback	callback The callback than handles error propagation or logging callbacks.
protected String	defaultTopicId The name of the default topic this producer is writing data to.
protected FlinkKafkaPartitioner<IN>	flinkKafkaPartitioner User-provided partitioner for assigning an object to a Kafka partition for each topic.
protected boolean	flushOnCheckpoint If true, the producer will wait until all outstanding records have been send to the broker.
static String	KEY_DISABLE_METRICS Configuration key for disabling the metrics reporting.
protected boolean	logFailuresOnly Flag indicating whether to accept failures (and log them), or to fail on failures.
protected long	pendingRecords Number of unacknowledged records.
protected SerializableObject	pendingRecordsLock Lock for accessing the pending records.
protected org.apache.kafka.clients.producer.KafkaProducer<byte[],byte[]>	producer KafkaProducer instance.
protected Properties	producerConfig User defined properties for the Producer.
protected KeyedSerializationSchema<IN>	schema (Serializable) SerializationSchema for turning objects used with Flink into.
protected Map<String,int[]>	topicPartitionsMap Partitions of each topic.

Constructor Summary

Constructors

Constructor and Description
FlinkKafkaProducerBase(String defaultTopicId, KeyedSerializationSchema<IN> serializationSchema, Properties producerConfig, FlinkKafkaPartitioner<IN> customPartitioner) The main constructor for creating a FlinkKafkaProducer.

Method Summary

Modifier and Type	Method and Description
protected void	checkErroneous()
void	close() Tear-down method for the user code.
protected abstract void	flush() Flush pending records.
protected <K,V> org.apache.kafka.clients.producer.KafkaProducer<K,V>	getKafkaProducer(Properties props) Used for testing only.
protected static int[]	getPartitionsByTopic(String topic, org.apache.kafka.clients.producer.KafkaProducer<byte[],byte[]> producer)
static Properties	getPropertiesFromBrokerList(String brokerList)
void	initializeState(FunctionInitializationContext context) This method is called when the parallel function instance is created during distributed execution.
void	invoke(IN next, SinkFunction.Context context) Called when new data arrives to the sink, and forwards it to Kafka.
protected long	numPendingRecords()
void	open(Configuration configuration) Initializes the connection to Kafka.
void	setFlushOnCheckpoint(boolean flush) If set to true, the Flink producer will wait for all outstanding messages in the Kafka buffers to be acknowledged by the Kafka producer on a checkpoint.
void	setLogFailuresOnly(boolean logFailuresOnly) Defines whether the producer should fail on errors, or only log them.
void	snapshotState(FunctionSnapshotContext ctx) This method is called when a snapshot for a checkpoint is requested.

Methods inherited from class org.apache.flink.api.common.functions.AbstractRichFunction

getIterationRuntimeContext, getRuntimeContext, setRuntimeContext

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.flink.streaming.api.functions.sink.SinkFunction

invoke

Field Detail

KEY_DISABLE_METRICS

public static final String KEY_DISABLE_METRICS

Configuration key for disabling the metrics reporting.

See Also:

Constant Field Values

producerConfig

protected final Properties producerConfig

User defined properties for the Producer.

defaultTopicId

protected final String defaultTopicId

The name of the default topic this producer is writing data to.

schema

protected final KeyedSerializationSchema<IN> schema

(Serializable) SerializationSchema for turning objects used with Flink into. byte[] for Kafka.

flinkKafkaPartitioner

protected final FlinkKafkaPartitioner<IN> flinkKafkaPartitioner

User-provided partitioner for assigning an object to a Kafka partition for each topic.

topicPartitionsMap

protected final Map<String,int[]> topicPartitionsMap

Partitions of each topic.

logFailuresOnly

protected boolean logFailuresOnly

Flag indicating whether to accept failures (and log them), or to fail on failures.

flushOnCheckpoint

protected boolean flushOnCheckpoint

If true, the producer will wait until all outstanding records have been send to the broker.

producer

protected transient org.apache.kafka.clients.producer.KafkaProducer<byte[],byte[]> producer

KafkaProducer instance.

callback

protected transient org.apache.kafka.clients.producer.Callback callback

The callback than handles error propagation or logging callbacks.

asyncException

protected transient volatile Exception asyncException

Errors encountered in the async producer are stored here.

pendingRecordsLock

protected final SerializableObject pendingRecordsLock

Lock for accessing the pending records.

pendingRecords

protected long pendingRecords

Number of unacknowledged records.

Constructor Detail

FlinkKafkaProducerBase

public FlinkKafkaProducerBase(String defaultTopicId,
                              KeyedSerializationSchema<IN> serializationSchema,
                              Properties producerConfig,
                              FlinkKafkaPartitioner<IN> customPartitioner)

The main constructor for creating a FlinkKafkaProducer.

Parameters:

defaultTopicId - The default topic to write data to

serializationSchema - A serializable serialization schema for turning user objects into a kafka-consumable byte[] supporting key/value messages

producerConfig - Configuration properties for the KafkaProducer. 'bootstrap.servers.' is the only required argument.

customPartitioner - A serializable partitioner for assigning messages to Kafka partitions. Passing null will use Kafka's partitioner.

Method Detail

setLogFailuresOnly

public void setLogFailuresOnly(boolean logFailuresOnly)

Defines whether the producer should fail on errors, or only log them. If this is set to true, then exceptions will be only logged, if set to false, exceptions will be eventually thrown and cause the streaming program to fail (and enter recovery).

Parameters:

logFailuresOnly - The flag to indicate logging-only on exceptions.

setFlushOnCheckpoint

public void setFlushOnCheckpoint(boolean flush)

If set to true, the Flink producer will wait for all outstanding messages in the Kafka buffers to be acknowledged by the Kafka producer on a checkpoint. This way, the producer can guarantee that messages in the Kafka buffers are part of the checkpoint.

Parameters:

flush - Flag indicating the flushing mode (true = flush on checkpoint)

getKafkaProducer

@VisibleForTesting
protected <K,V> org.apache.kafka.clients.producer.KafkaProducer<K,V> getKafkaProducer(Properties props)

Used for testing only.

open

public void open(Configuration configuration)
          throws Exception

Initializes the connection to Kafka.

Specified by:

open in interface RichFunction

Overrides:

open in class AbstractRichFunction

Parameters:

configuration - The configuration containing the parameters attached to the contract.

Throws:

Exception - Implementations may forward exceptions, which are caught by the runtime. When the runtime catches an exception, it aborts the task and lets the fail-over logic decide whether to retry the task execution.

See Also:

Configuration

invoke

public void invoke(IN next,
                   SinkFunction.Context context)
            throws Exception

Called when new data arrives to the sink, and forwards it to Kafka.

Specified by:

invoke in interface SinkFunction<IN>

Parameters:

next - The incoming data

context - Additional context about the input record.

Throws:

Exception - This method may throw exceptions. Throwing an exception will cause the operation to fail and may trigger recovery.

close

public void close()
           throws Exception

Description copied from interface: RichFunction

Tear-down method for the user code. It is called after the last call to the main working methods (e.g. map or join). For functions that are part of an iteration, this method will be invoked after each iteration superstep.

This method can be used for clean up work.

Specified by:

close in interface RichFunction

Overrides:

close in class AbstractRichFunction

Throws:

Exception - Implementations may forward exceptions, which are caught by the runtime. When the runtime catches an exception, it aborts the task and lets the fail-over logic decide whether to retry the task execution.

flush

protected abstract void flush()

Flush pending records.

initializeState

public void initializeState(FunctionInitializationContext context)
                     throws Exception

Description copied from interface: CheckpointedFunction

This method is called when the parallel function instance is created during distributed execution. Functions typically set up their state storing data structures in this method.

Specified by:

initializeState in interface CheckpointedFunction

Parameters:

context - the context for initializing the operator

Throws:

Exception - Thrown, if state could not be created ot restored.

snapshotState

public void snapshotState(FunctionSnapshotContext ctx)
                   throws Exception

Description copied from interface: CheckpointedFunction

This method is called when a snapshot for a checkpoint is requested. This acts as a hook to the function to ensure that all state is exposed by means previously offered through FunctionInitializationContext when the Function was initialized, or offered now by FunctionSnapshotContext itself.

Specified by:

snapshotState in interface CheckpointedFunction

Parameters:

ctx - the context for drawing a snapshot of the operator

Throws:

Exception - Thrown, if state could not be created ot restored.

checkErroneous

protected void checkErroneous()
                       throws Exception

Throws:

Exception

getPropertiesFromBrokerList

public static Properties getPropertiesFromBrokerList(String brokerList)

getPartitionsByTopic

protected static int[] getPartitionsByTopic(String topic,
                                            org.apache.kafka.clients.producer.KafkaProducer<byte[],byte[]> producer)

numPendingRecords

@VisibleForTesting
protected long numPendingRecords()