WIP: txnmgr

(cherry picked from commit 880fca6d0853b8b947ad00bafd206105080a0309)

Author: edenhill

CONFIGURATION.md

@@ -113,6 +113,8 @@ rebalance_cb                             |  C  |                 |
 offset_commit_cb                         |  C  |                 |               | low        | Offset commit result propagation callback. (set with rd_kafka_conf_set_offset_commit_cb()) <br>*Type: pointer*
 enable.partition.eof                     |  C  | true, false     |         false | low        | Emit RD_KAFKA_RESP_ERR__PARTITION_EOF event whenever the consumer reaches the end of a partition. <br>*Type: boolean*
 check.crcs                               |  C  | true, false     |         false | medium     | Verify CRC32 of consumed messages, ensuring no on-the-wire or on-disk corruption to the messages occurred. This check comes at slightly increased CPU usage. <br>*Type: boolean*
+transactional.id                         |  P  |                 |               | high       | The TransactionalId to use for transactional delivery. This enables reliability semantics which span multiple producer sessions since it allows the client to guarantee that transactions using the same TransactionalId have been completed prior to starting any new transactions. If no TransactionalId is provided, then the producer is limited to idempotent delivery. Note that enable.idempotence must be enabled if a TransactionalId is configured. The default is null, which means transactions cannot be used. Note that, by default, transactions require a cluster of at least three brokers which is the recommended setting for production; for development you can change this, by adjusting broker setting transaction.state.log.replication.factor. <br>*Type: string*
+transaction.timeout.ms                   |  P  | 1000 .. 2147483647 |         60000 | medium     | The maximum amount of time in milliseconds that the transaction coordinator will wait for a transaction status update from the producer before proactively aborting the ongoing transaction. If this value is larger than the `transaction.max.timeout.ms` setting in the broker, the init_transactions() call will fail with ERR_INVALID_TRANSACTION_TIMEOUT. The transaction timeout must be at least 1000 ms larger than `message.timeout.ms` and `socket.timeout.ms`. <br>*Type: integer*
 enable.idempotence                       |  P  | true, false     |         false | high       | When set to `true`, the producer will ensure that messages are successfully produced exactly once and in the original produce order. The following configuration properties are adjusted automatically (if not modified by the user) when idempotence is enabled: `max.in.flight.requests.per.connection=5` (must be less than or equal to 5), `retries=INT32_MAX` (must be greater than 0), `acks=all`, `queuing.strategy=fifo`. Producer instantation will fail if user-supplied configuration is incompatible. <br>*Type: boolean*
 enable.gapless.guarantee                 |  P  | true, false     |         false | low        | **EXPERIMENTAL**: subject to change or removal. When set to `true`, any error that could result in a gap in the produced message series when a batch of messages fails, will raise a fatal error (ERR__GAPLESS_GUARANTEE) and stop the producer. Messages failing due to `message.timeout.ms` are not covered by this guarantee. Requires `enable.idempotence=true`. <br>*Type: boolean*
 queue.buffering.max.messages             |  P  | 1 .. 10000000   |        100000 | high       | Maximum number of messages allowed on the producer queue. This queue is shared by all topics and partitions. <br>*Type: integer*

INTRODUCTION.md

@@ -701,15 +701,106 @@ which returns one of the following values:
 This method should be called by the application on delivery report error.
 
 
+### Transactional Producer
 
+#### FIXME: misc
+
+To make sure messages time out (in case of connectivity problems, etc) within
+the transaction, the `message.timeout.ms` configuration property must be
+set lower than the `transaction.timeout.ms`.
+If `message.timeout.ms` is not explicitly configured it will be adjusted
+automatically.
+
+
+#### Old producer fencing
+
+If a new transactional producer instance is started with the same
+`transactional.id`, any previous still running producer
+instance will be fenced off at the next produce, commit or abort attempt, by
+raising a fatal error with the error code set to
+`RD_KAFKA_RESP_ERR__FENCED`.
+
+
+
+### Exactly Once Semantics (EOS) and transactions
+
+librdkafka supports Exactly One Semantics (EOS) as defined in [KIP-98](https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging).
+For more on the use of transactions, see [Transactions in Apache Kafka](https://www.confluent.io/blog/transactions-apache-kafka/).
+
+
+The transactional consume-process-produce loop roughly boils down to the
+following pseudo-code:
+
+```c
+    /* Producer */
+    rd_kafka_conf_t *pconf = rd_kafka_conf_new();
+    rd_kafka_conf_set(pconf, "bootstrap.servers", "mybroker");
+    rd_kafka_conf_set(pconf, "transactional.id", "my-transactional-id");
+    rd_kafka_t *producer = rd_kafka_new(RD_KAFKA_PRODUCER, pconf);
+
+    rd_kafka_init_transactions(producer);
+
+
+    /* Consumer */
+    rd_kafka_conf_t *cconf = rd_kafka_conf_new();
+    rd_kafka_conf_set(cconf, "bootstrap.servers", "mybroker");
+    rd_kafka_conf_set(cconf, "group.id", "my-group-id");
+    rd_kafka_conf_set(cconf, "enable.auto.commit", "false");
+    rd_kafka_t *consumer = rd_kafka_new(RD_KAFKA_CONSUMER, cconf);
+    rd_kafka_poll_set_consumer(consumer);
+
+    rd_kafka_subscribe(consumer, "inputTopic");
+
+    /* Consume-Process-Produce loop */
+    while (run) {
+
+       /* Begin transaction */
+       rd_kafka_begin_transaction(producer);
+
+       while (some_limiting_factor) {
+           rd_kafka_message_t *in, *out;
+
+           /* Consume messages */
+           in = rd_kafka_consumer_poll(consumer, -1);
+
+           /* Process message, generating an output message */
+           out = process_msg(in);
+
+           /* Produce output message to output topic */
+           rd_kafka_produce(producer, "outputTopic", out);
+
+           / FIXME: or perhaps */
+           rd_kafka_topic_partition_list_set_from_msg(processed, msg);
+           /* or */
+           rd_kafka_transaction_store_offset_from_msg(producer, msg);
+       }
+
+       /* Commit the consumer offset as part of the transaction */
+       rd_kafka_send_offsets_to_transaction(producer,
+                                            "my-group-id",
+                                            rd_kafka_position(consumer));
+                                            /* or processed */
+
+       /* Commit the transaction */
+       rd_kafka_commit_transaction(producer);
+   }
+
+   rd_kafka_consumer_close(consumer);
+   rd_kafka_destroy(consumer);
+   rd_kafka_destroy(producer);
+```
+
+**Note**: The above code is a logical representation of transactional
+          program flow and does not represent the exact API parameter usage.
+          A proper application will perform error handling, etc.
+          See [`examples/transactions.cpp`](examples/transactions.cpp) for a proper example.
 
 
 ## Usage
 
 ### Documentation
 
-The librdkafka API is documented in the
-[`rdkafka.h`](src/rdkafka.h)
+The librdkafka API is documented in the [`rdkafka.h`](src/rdkafka.h)
 header file, the configuration properties are documented in
 [`CONFIGURATION.md`](CONFIGURATION.md)
 
@@ -777,10 +868,18 @@ Configuration is applied prior to object creation using the
         rd_kafka_conf_destroy(rk);
         fail("Failed to create producer: %s\n", errstr);
     }
-    
+
     /* Note: librdkafka takes ownership of the conf object on success */
 ```
 
+Configuration properties may be set in any order (except for interceptors) and
+may be overwritten before being passed to `rd_kafka_new()`.
+`rd_kafka_new()` will verify that the passed configuration is consistent
+and will fail and return an error if incompatible configuration properties
+are detected. It will also emit log warnings for deprecated and problematic
+configuration properties.
+
+
 ### Termination
 
 librdkafka is asynchronous in its nature and performs most operation in its

src-cpp/rdkafkacpp.h

@@ -279,6 +279,12 @@ enum ErrorCode {
         ERR__GAPLESS_GUARANTEE = -148,
         /** Maximum poll interval exceeded */
         ERR__MAX_POLL_EXCEEDED = -147,
+        /** Unknown broker */
+        ERR__UNKNOWN_BROKER = -146,
+        /** Functionality not configured */
+        ERR__NOT_CONFIGURED = -145,
+        /** Instance has been fenced */
+        ERR__FENCED = -144,
 
         /** End internal error codes */
 	ERR__END = -100,
@@ -2915,6 +2921,48 @@ class RD_EXPORT Producer : public virtual Handle {
                               * purging to finish. */
   };
 
+  /**
+   * Transactional API
+   *
+   * Requires Kafka broker version v0.11.0 or later
+   */
+
+  /**
+   * @brief
+   *
+   * @fixme blocking?
+   */
+  virtual ErrorCode init_transactions (int timeout_ms,
+                                       std::string &errstr) = 0;
+
+  /**
+   * @fixme blocking?
+   */
+  virtual ErrorCode begin_transaction (std::string &errstr) = 0;
+
+  /**
+   * @fixme blocking?
+   */
+  virtual ErrorCode send_offsets_to_transaction (
+          const std::vector<TopicPartition*> &offsets,
+          const std::string &group_id,
+          std::string &errstr) = 0;
+
+  /**
+   * @brief
+   *
+   * @fixme blocking?
+   */
+  virtual ErrorCode commit_transaction (int timeout_ms,
+                                        std::string &errstr) = 0;
+
+  /**
+   * @brief
+   *
+   * @fixme blocking?
+   */
+  virtual ErrorCode abort_transaction (int timeout_ms,
+                                       std::string &errstr) = 0;
 };
 
 /**@}*/

src-cpp/rdkafkacpp_int.h

@@ -1241,6 +1241,75 @@ class ProducerImpl : virtual public Producer, virtual public HandleImpl {
                                                                 (int)purge_flags));
   }
 
+  ErrorCode init_transactions (int timeout_ms, std::string &errstr) {
+    rd_kafka_resp_err_t c_err;
+    char errbuf[512];
+
+    c_err = rd_kafka_init_transactions(rk_, timeout_ms,
+                                       errbuf, sizeof(errbuf));
+    if (c_err)
+      errstr = errbuf;
+
+    return static_cast<ErrorCode>(c_err);
+  }
+
+  ErrorCode begin_transaction (std::string &errstr) {
+    rd_kafka_resp_err_t c_err;
+    char errbuf[512];
+
+    c_err = rd_kafka_begin_transaction(rk_, errbuf, sizeof(errbuf));
+    if (c_err)
+      errstr = errbuf;
+
+    return static_cast<ErrorCode>(c_err);
+  }
+
+  ErrorCode send_offsets_to_transaction (
+      const std::vector<TopicPartition*> &offsets,
+      const std::string &group_id,
+      std::string &errstr) {
+    rd_kafka_resp_err_t c_err;
+    char errbuf[512];
+    rd_kafka_topic_partition_list_t *c_offsets = partitions_to_c_parts(offsets);
+
+    c_err = rd_kafka_send_offsets_to_transaction(rk_, c_offsets,
+                                                 group_id.c_str(),
+                                                 errbuf, sizeof(errbuf));
+
+    rd_kafka_topic_partition_list_destroy(c_offsets);
+
+    if (c_err)
+      errstr = errbuf;
+
+    return static_cast<ErrorCode>(c_err);
+
+  }
+
+  ErrorCode commit_transaction (int timeout_ms, std::string &errstr) {
+    rd_kafka_resp_err_t c_err;
+    char errbuf[512];
+
+    c_err = rd_kafka_commit_transaction(rk_, timeout_ms,
+                                        errbuf, sizeof(errbuf));
+    if (c_err)
+      errstr = errbuf;
+
+    return static_cast<ErrorCode>(c_err);
+
+  }
+
+  ErrorCode abort_transaction (int timeout_ms, std::string &errstr) {
+    rd_kafka_resp_err_t c_err;
+    char errbuf[512];
+
+    c_err = rd_kafka_abort_transaction(rk_, timeout_ms, errbuf, sizeof(errbuf));
+    if (c_err)
+      errstr = errbuf;
+
+    return static_cast<ErrorCode>(c_err);
+
+  }
+
   static Producer *create (Conf *conf, std::string &errstr);
 
 };

src/CMakeLists.txt

@@ -42,6 +42,7 @@ set(
     rdkafka_background.c
     rdkafka_idempotence.c
     rdkafka_cert.c
+    rdkafka_coord.c
     rdkafka_mock.c
     rdkafka_mock_handlers.c
     rdlist.c

src/Makefile

@@ -47,6 +47,7 @@ SRCS=		rdkafka.c rdkafka_broker.c rdkafka_msg.c rdkafka_topic.c \
 		rdkafka_msgset_writer.c rdkafka_msgset_reader.c \
 		rdkafka_header.c rdkafka_admin.c rdkafka_aux.c \
 		rdkafka_background.c rdkafka_idempotence.c rdkafka_cert.c \
+		rdkafka_txnmgr.c rdkafka_coord.c \
 		rdvarint.c rdbuf.c rdunittest.c \
 		rdkafka_mock.c rdkafka_mock_handlers.c \
 		$(SRCS_y)

src/rd.h

@@ -344,20 +344,21 @@ static RD_INLINE RD_UNUSED int rd_refcnt_get (rd_refcnt_t *R) {
         } while (0)
 
 #if ENABLE_REFCNT_DEBUG
-#define rd_refcnt_add(R)                                                \
+#define rd_refcnt_add_fl(FUNC,LINE,R)                                   \
         (                                                               \
                 printf("REFCNT DEBUG: %-35s %d +1: %16p: %s:%d\n",      \
-                       #R, rd_refcnt_get(R), (R), __FUNCTION__,__LINE__), \
+                       #R, rd_refcnt_get(R), (R), (FUNC), (LINE)),      \
                 rd_refcnt_add0(R)                                       \
                 )
 
+#define rd_refcnt_add(R) rd_refcnt_add_fl(__FUNCTION__, __LINE__, (R))
+
 #define rd_refcnt_add2(R,WHAT)  do {                                        \
                 printf("REFCNT DEBUG: %-35s %d +1: %16p: %16s: %s:%d\n",      \
                        #R, rd_refcnt_get(R), (R), WHAT, __FUNCTION__,__LINE__), \
                 rd_refcnt_add0(R);                                      \
         } while (0)
 
-
 #define rd_refcnt_sub2(R,WHAT) (                                            \
                 printf("REFCNT DEBUG: %-35s %d -1: %16p: %16s: %s:%d\n",      \
                        #R, rd_refcnt_get(R), (R), WHAT, __FUNCTION__,__LINE__), \
@@ -369,6 +370,7 @@ static RD_INLINE RD_UNUSED int rd_refcnt_get (rd_refcnt_t *R) {
                 rd_refcnt_sub0(R) )
 
 #else
+#define rd_refcnt_add_fl(FUNC,LINE,R)  rd_refcnt_add0(R)
 #define rd_refcnt_add(R)  rd_refcnt_add0(R)
 #define rd_refcnt_sub(R)  rd_refcnt_sub0(R)
 #endif