* Properties with random string as default

* `client.id`

* `group.id`

* Different default value from ***librdkafka***

* KafkaSyncProducer

* `linger.ms` (default: `0`)

You should not use a `KafkaSyncProducer` if you want to get a high throughput. The `Sync` means the `send` operation is a synchronous operation, and would not go on until the `acks` are received.

* It means the delay in milliseconds to wait for messages in the producer queue to accumulate before constructing MessageSets to transmit to brokers.

1. For a `KafkaSyncProducer`, it sends the messages one by one (after `acks` response received), thus it could use `linger.ms=0` (as default) to eliminate the unnecessary waiting time.

2. For a `KafkaAsyncProduer`, a larger `linger.ms` could be used to accumulate messages for MessageSets to improve the performance -- it should be the result of balancing between throughput and latency.

* Default queue.buffing setting,

* `queue.buffering.max.messages=1000000`

* `queue.buffering.max.kbytes=0x100000` (1GB)

* Normally, the default settings should be good enough.

* Take an example, -- a `KafkaAsyncProducer` just sent many messages, and a few of these messages (in the middle) failed to be delivered successfully. While the producer got the sending error, it might retry sending these messages again, thus causing the disordering.

* To avoid it. Two options,

* Use a `KafkaSyncProducer`, but this would severely impact the throughput.

* Embed some `sequence number` (e.g, record id, part of the `key`, etc) in the `ProducerRecord`, for de-duplication.

* The `enable.idempotence` configuration would resolve such a problem. And this configuration is RECOMMENDED for both `KafkaSyncProducer` and `KafkaAsyncProducer`, as long as it's possible.

int ret = 0;

KafkaSyncProducer producer(

Properties({

{ ProducerConfig::BOOTSTRAP_SERVERS, "192.168.0.1:9092,192.168.0.2:9092,192.168.0.3:9092" },

{ ProducerConfig::ENABLE_IDEMPOTENCE, "true" },

{ ProducerConfig::MESSAGE_TIMEOUT_MS, "60000"}

})

);

while (running) {

const auto& msg = topMsgOfUpstream();

try {

auto record = ProducerRecord(topic, msg.key, msg.value);

producer.send(record);

popMsgFromUpstream();

} catch (const KafkaException& e) {

std::cerr << "Failed to send message! Reason: " << e.what() << std::endl;

ret = e.error();

break;

}

}

producer.close();

return ret;

Unlike Java's KafkaProducer, here we introduce two derived classes, -- `KafkaSyncProducer` and `KafkaAsyncProducer` --depending on different `send` behaviors (synchronous/asynchronous).

* The "Sync" (in the name) means `send` is a blocking operation, and it will immediately get the RecordMetadata while the function returns. If anything wrong occurs, an exception would be thrown.

// Create configuration object

kafka::Properties props({

{"bootstrap.servers", brokers},

{"enable.idempotence", "true"},

});

// Create a producer instance.

kafka::KafkaSyncProducer producer(props);

// Read messages from stdin and produce to the broker.

std::cout << "% Type message value and hit enter to produce message. (empty line to quit)" << std::endl;

for (std::string line; std::getline(std::cin, line);) {

// The ProducerRecord doesn't own `line`, it is just a thin wrapper

auto record = kafka::ProducerRecord(topic,

kafka::NullKey,

kafka::Value(line.c_str(), line.size()));

// Send the message.

try {

kafka::Producer::RecordMetadata metadata = producer.send(record);

std::cout << "% Message delivered: " << metadata.toString() << std::endl;

} catch (const kafka::KafkaException& e) {

std::cerr << "% Message delivery failed: " << e.error().message() << std::endl;

}

if (line.empty()) break;

};

// producer.close(); // No explicit close is needed, RAII will take care of it

* Same with KafkaSyncProducer, the user must guarantee the memory block for `ProducerRecord`'s `key` is valid until being `send`.

While we construct a `KafkaAsyncProducer` with option `KafkaClient::EventsPollingOption::Auto` (default), an internal thread would be created for `MessageDelivery` callbacks handling.

* Note, if you constructed the `KafkaAsyncProducer` with `EventsPollingOption::Manual`, the `send()` would be an `unblocked` operation.

kafak::KafkaAsyncProducer producer(props, KafkaClient::EventsPollingOption::Manual);

1. `KafkaSyncProducer` gets `std::error_code` by catching exceptions (with `error()` member function).

2. `KafkaAsyncProducer` gets `std::error_code` with delivery-callback (with a parameter of the callback function).

If an ack failed to be received within `MESSAGE_TIMEOUT_MS`, an exception would be thrown for a KafkaSyncSend, or, an error code would be received by the delivery callback for a KafkaAsyncProducer.

2. Use a `KafkaSyncProducer` (with configuration `{ProducerConfig::ACKS, "all"}`); or use a `KafkaAsyncProducer` (with configuration `{ProducerConfig::ENABLE_IDEMPOTENCE, "true"}`), together with proper error-handling within the delivery callbacks.

Excluding the user's main thread, KafkaSyncProducer would start another (N + 2) threads in the background, while `KafkaAsyncProducer` would start (N + 3) background threads. (N means the number of BOOTSTRAP_SERVERS)

3. `KafkaAsyncProducer` has one more background thread to keep polling the delivery callback event.

E.g, if a KafkaSyncProducer was created with property of `BOOTSTRAP_SERVERS=127.0.0.1:8888,127.0.0.1:8889,127.0.0.1:8890`, it would take 6 threads in total (including the main thread).

The `Producer::Callback` is only available for a `KafkaAsyncProducer`.

Note, should be careful if both the `KafkaAsyncProducer::send()` and the `Producer::Callback` might access the same container at the same time.

kafka::KafkaAsyncProducer producer(props);

kafka::KafkaAsyncProducer producer(props);

kafka::KafkaSyncProducer producer(props);

kafka::Producer::RecordMetadata metadata = producer.send(record);