org.apache.flink.streaming.connectors.kafka.internal

Class FlinkKafkaInternalProducer<K,V>

java.lang.Object

org.apache.flink.streaming.connectors.kafka.internal.FlinkKafkaInternalProducer<K,V>

All Implemented Interfaces:

Closeable, AutoCloseable, org.apache.kafka.clients.producer.Producer<K,V>

@PublicEvolving
public class FlinkKafkaInternalProducer<K,V>
extends Object
implements org.apache.kafka.clients.producer.Producer<K,V>

Internal flink kafka producer.

Field Summary

Fields

Modifier and Type	Field and Description
protected org.apache.kafka.clients.producer.KafkaProducer<K,V>	kafkaProducer
protected String	transactionalId

Constructor Summary

Constructors

Constructor and Description
FlinkKafkaInternalProducer(Properties properties)

Method Summary

Modifier and Type	Method and Description
void	abortTransaction()
void	beginTransaction()
void	close()
void	close(long timeout, TimeUnit unit)
void	commitTransaction()
void	flush()
protected static Enum<?>	getEnum(String enumFullName)
short	getEpoch()
long	getProducerId()
String	getTransactionalId()
int	getTransactionCoordinatorId()
protected static Object	getValue(Object object, String fieldName)
void	initTransactions()
protected static Object	invoke(Object object, String methodName, Object... args)
Map<org.apache.kafka.common.MetricName,? extends org.apache.kafka.common.Metric>	metrics()
List<org.apache.kafka.common.PartitionInfo>	partitionsFor(String topic)
void	resumeTransaction(long producerId, short epoch) Instead of obtaining producerId and epoch from the transaction coordinator, re-use previously obtained ones, so that we can resume transaction after a restart.
Future<org.apache.kafka.clients.producer.RecordMetadata>	send(org.apache.kafka.clients.producer.ProducerRecord<K,V> record)
Future<org.apache.kafka.clients.producer.RecordMetadata>	send(org.apache.kafka.clients.producer.ProducerRecord<K,V> record, org.apache.kafka.clients.producer.Callback callback)
void	sendOffsetsToTransaction(Map<org.apache.kafka.common.TopicPartition,org.apache.kafka.clients.consumer.OffsetAndMetadata> offsets, String consumerGroupId)
protected static void	setValue(Object object, String fieldName, Object value)

Methods inherited from class java.lang.Object

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

Field Detail

kafkaProducer

protected final org.apache.kafka.clients.producer.KafkaProducer<K,V> kafkaProducer

transactionalId

@Nullable
protected final String transactionalId

Constructor Detail

FlinkKafkaInternalProducer

public FlinkKafkaInternalProducer(Properties properties)

Method Detail

initTransactions

public void initTransactions()

beginTransaction

public void beginTransaction()
                      throws org.apache.kafka.common.errors.ProducerFencedException

Throws:

org.apache.kafka.common.errors.ProducerFencedException

commitTransaction

public void commitTransaction()
                       throws org.apache.kafka.common.errors.ProducerFencedException

Throws:

org.apache.kafka.common.errors.ProducerFencedException

abortTransaction

public void abortTransaction()
                      throws org.apache.kafka.common.errors.ProducerFencedException

Throws:

org.apache.kafka.common.errors.ProducerFencedException

sendOffsetsToTransaction

public void sendOffsetsToTransaction(Map<org.apache.kafka.common.TopicPartition,org.apache.kafka.clients.consumer.OffsetAndMetadata> offsets,
                                     String consumerGroupId)
                              throws org.apache.kafka.common.errors.ProducerFencedException

Throws:

org.apache.kafka.common.errors.ProducerFencedException

send

public Future<org.apache.kafka.clients.producer.RecordMetadata> send(org.apache.kafka.clients.producer.ProducerRecord<K,V> record)

Specified by:

send in interface org.apache.kafka.clients.producer.Producer<K,V>

send

public Future<org.apache.kafka.clients.producer.RecordMetadata> send(org.apache.kafka.clients.producer.ProducerRecord<K,V> record,
                                                                     org.apache.kafka.clients.producer.Callback callback)

Specified by:

send in interface org.apache.kafka.clients.producer.Producer<K,V>

partitionsFor

public List<org.apache.kafka.common.PartitionInfo> partitionsFor(String topic)

Specified by:

partitionsFor in interface org.apache.kafka.clients.producer.Producer<K,V>

metrics

public Map<org.apache.kafka.common.MetricName,? extends org.apache.kafka.common.Metric> metrics()

Specified by:

metrics in interface org.apache.kafka.clients.producer.Producer<K,V>

close

public void close()

Specified by:

close in interface Closeable

Specified by:

close in interface AutoCloseable

Specified by:

close in interface org.apache.kafka.clients.producer.Producer<K,V>

close

public void close(long timeout,
                  TimeUnit unit)

Specified by:

close in interface org.apache.kafka.clients.producer.Producer<K,V>

flush

public void flush()

Specified by:

flush in interface org.apache.kafka.clients.producer.Producer<K,V>

resumeTransaction

public void resumeTransaction(long producerId,
                              short epoch)

Instead of obtaining producerId and epoch from the transaction coordinator, re-use previously obtained ones, so that we can resume transaction after a restart. Implementation of this method is based on KafkaProducer#initTransactions. https://github.com/apache/kafka/commit/5d2422258cb975a137a42a4e08f03573c49a387e#diff-f4ef1afd8792cd2a2e9069cd7ddea630

getTransactionalId

public String getTransactionalId()

getProducerId

public long getProducerId()

getEpoch

public short getEpoch()

getTransactionCoordinatorId

@VisibleForTesting
public int getTransactionCoordinatorId()

getEnum

protected static Enum<?> getEnum(String enumFullName)

invoke

protected static Object invoke(Object object,
                               String methodName,
                               Object... args)

getValue

protected static Object getValue(Object object,
                                 String fieldName)

setValue

protected static void setValue(Object object,
                               String fieldName,
                               Object value)