How to send messages synchronously in kafka?

Admin 1
Viewed 24886

One way of achieving it could be by setting the properties parameter
max.in.flight.requests.per.connection = 1.

But I want to know if there is an even direct or alternate way of sending messages synchronously in kafka, something like producer.syncSend(...).

kafka-producer-api apache-kafka
5 Answers

As Thilo suggested, you can call Future#get to block until the sending has completed. However you might have some performance issue, since the producer starts sending when the producer queue has batch.size elements, when the buffer of size buffer.memory is full or after max.block.ms milliseconds.

If you have a limited number of threads pushing to kafka, you will have to wait max.block.ms each time for your message to be sent. So in some cases, you will prefer using :

// send message to producer queue
Future<RecordMetadata> future = producer.send(new ProducerRecord<>(topic, key, message));
// flush producer queue to spare queuing time
producer.flush();
// throw error when kafka is unreachable
future.get(10, TimeUnit.SECONDS);
user_17165700

From my adventures with Kafka :-) order of message production can only be guaranteed if you have one Producer thread and set max.in.flight.requests.per.connection = 1 (or turn of retries, i.e. retries= 0 or both).

If you what to scale to more than one Producer, then you have to "make sure" that messages that will be stored to the same partition will be produced by the same Producer instance.

user_37947440

When max.in.flight.requests.per.connection = 1, it just means the ordering of messages is guaranteed within a partition it has nothing to do with synchronization.

Python code in-case. For a synchronous send, make sure to block on the future with a good time-out.

from kafka import KafkaProducer
from kafka.errors import KafkaError

#by default ack = 1, if ack = 'all' --> waits for acks from replicas 
producer = KafkaProducer(bootstrap_servers=['brokerIP:9092'], ack= 'all')


key = b'key'
value = b'value'

future = producer.send("my-topic", key=key, value=value)

# block on this future for sync sends
try:
    record_metadata = future.get(timeout=10)
except KafkaError:
    log.exception()
    pass

print (record_metadata.topic)
print (record_metadata.partition)
print (record_metadata.offset)

producer.flush()
producer.close()
user_38858190
Related