Do newer versions of Kafka producers still have "producer.type"?

Question

Older versions' doc says it's one of the essential properties.

Newer versions' doc doesn't mention it at all.

Do newer versions of Kafka producers still have producer.type?

Or, new producers are always async, and I should call future.get() to make it sync?

Answer 1

New producers are always async, and you should call future.get() to make it sync. It's not worth making two apis methods when something as simple as adding future.get() gives you basically the same functionality.

From the documentation for send() here

https://kafka.apache.org/0110/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html

Since the send call is asynchronous it returns a Future for the RecordMetadata that will be assigned to this record. Invoking get() on this future will block until the associated request completes and then return the metadata for the record or throw any exception that occurred while sending the record.

If you want to simulate a simple blocking call you can call the get() method immediately:

byte[] key = "key".getBytes();
byte[] value = "value".getBytes();  
ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("my-topic", key, value);
producer.send(record).get();

Answer 2

Why do you want to make the send() to sync ?

This is a kafka feature to batch message for better throughput.

Asynchronous send

Batching is one of the big drivers of efficiency, and to enable batching the Kafka producer will attempt to accumulate data in memory and to send out larger batches in a single request. The batching can be configured to accumulate no more than a fixed number of messages and to wait no longer than some fixed latency bound (say 64k or 10 ms). This allows the accumulation of more bytes to send, and few larger I/O operations on the servers. This buffering is configurable and gives a mechanism to trade off a small amount of additional latency for better throughput.

There is no way to do a send sync because of the api only support the async method, But there is a some configs you can specify to do some work arround.

You could set the batch.size to 0. In this case, the message bacthing is disabled.

However I think you should just leave the batch.size default and set the linger.ms to 0 (this is also default). In this case, if many message come in the same time, they will be batched in one send immediately .

The producer groups together any records that arrive in between request transmissions into a single batched request. Normally this occurs only under load when records arrive faster than they can be sent out.

And if you want to make sure the message is sent and persisted successfully, you coould set the acks to -1 or 1 and retries to 3 (e.g.)

More info about the producer config, you can refer https://kafka.apache.org/documentation/#producerconfigs