Applied most review comments

confluentinc · Nov 26, 2018 · 9821a3d · 9821a3d
1 parent c44fd07
commit 9821a3d
Show file tree

Hide file tree

Showing 11 changed files with 531 additions and 209 deletions.
diff --git a/examples/kafkatest_verifiable_client.cpp b/examples/kafkatest_verifiable_client.cpp
@@ -457,17 +457,6 @@ void msg_consume(RdKafka::KafkaConsumer *consumer,
               " [" << (int)msg->partition() << "]  at offset " <<
               msg->offset() << std::endl;
 
-        RdKafka::Headers *headers = msg->get_headers();
-        if (headers) {
-          std::vector<RdKafka::Headers::Header> sheaders = headers->get_all();
-          std::cout << "Headers length: " << sheaders.size() << std::endl;
-          for(std::vector<RdKafka::Headers::Header>::const_iterator it = sheaders.begin();
-              it != sheaders.end();
-              it++) {
-            std::cout << "Key: " << (*it).key << " Value: " << (*it).value << std::endl;
-          }
-        }
-
         if (state.maxMessages >= 0 &&
             state.consumer.consumedMessages >= state.maxMessages)
           return;
@@ -842,42 +831,31 @@ int main (int argc, char **argv) {
       msg << value_prefix << i;
       while (true) {
         RdKafka::ErrorCode resp;
-        RdKafka::Headers *headers = 0;
-        if (create_time == -1) {
-          resp = producer->produce(topic, partition,
-                                    RdKafka::Producer::RK_MSG_COPY /* Copy payload */,
-                                    const_cast<char *>(msg.str().c_str()),
-                                    msg.str().size(), NULL, NULL);
-        } else {
-          std::string name = "kafkaheader";
-          std::string val = "header_val";
-          std::vector<RdKafka::Headers::Header> headers_arr;
-          headers_arr.push_back(RdKafka::Headers::Header(name, val.c_str()));
-
-          headers = RdKafka::Headers::create(headers_arr, false);
-          resp = producer->produce(topics[0], partition,
+       if (create_time == -1) {
+         resp = producer->produce(topic, partition,
+                                  RdKafka::Producer::RK_MSG_COPY /* Copy payload */,
+                                  const_cast<char *>(msg.str().c_str()),
+                                  msg.str().size(), NULL, NULL);
+       } else {
+         resp = producer->produce(topics[0], partition,
                                   RdKafka::Producer::RK_MSG_COPY /* Copy payload */,
                                   const_cast<char *>(msg.str().c_str()),
                                   msg.str().size(),
                                   NULL, 0,
                                   create_time,
-                                  NULL, headers);
-        }
+                                  NULL);
+       }
 
         if (resp == RdKafka::ERR__QUEUE_FULL) {
           producer->poll(100);
           continue;
         } else if (resp != RdKafka::ERR_NO_ERROR) {
-          headers->destroy_headers();
           errorString("producer_send_error",
                       RdKafka::err2str(resp), topic->name(), NULL, msg.str());
           state.producer.numErr++;
         } else {
           state.producer.numSent++;
         }
-        if (headers) {
-          delete headers;
-        }
         break;
       }
 

diff --git a/src-cpp/HeadersImpl.cpp b/src-cpp/HeadersImpl.cpp
@@ -33,15 +33,15 @@
 
 #include "rdkafkacpp_int.h"
 
-RdKafka::Headers  *RdKafka::Headers::create(size_t initial_count, bool free_rd_headers) {
-    return new RdKafka::HeadersImpl(initial_count, free_rd_headers);
+RdKafka::Headers *RdKafka::Headers::create(size_t initial_count) {
+    return new RdKafka::HeadersImpl(initial_count, false);
 }
 
-RdKafka::Headers  *RdKafka::Headers::create(const std::vector<Header> &headers, bool free_rd_headers) {
+RdKafka::Headers *RdKafka::Headers::create(const std::vector<Header> &headers) {
     if (headers.size() > 0) {
-        return new RdKafka::HeadersImpl(headers, free_rd_headers);
+        return new RdKafka::HeadersImpl(headers, false);
     } else {
-        return 0;
+        return new RdKafka::HeadersImpl(8, false);
     }
 
 }

diff --git a/src-cpp/ProducerImpl.cpp b/src-cpp/ProducerImpl.cpp
@@ -143,19 +143,39 @@ RdKafka::ProducerImpl::produce (RdKafka::Topic *topic,
 
 }
 
+RdKafka::ErrorCode
+RdKafka::ProducerImpl::produce (const std::string topic_name,
+                                int32_t partition, int msgflags,
+                                void *payload, size_t len,
+                                const void *key, size_t key_len,
+                                int64_t timestamp, void *msg_opaque) {
+  return
+  static_cast<RdKafka::ErrorCode>
+    (
+      rd_kafka_producev(rk_,
+                        RD_KAFKA_V_TOPIC(topic_name.c_str()),
+                        RD_KAFKA_V_PARTITION(partition),
+                        RD_KAFKA_V_MSGFLAGS(msgflags),
+                        RD_KAFKA_V_VALUE(payload, len),
+                        RD_KAFKA_V_KEY(key, key_len),
+                        RD_KAFKA_V_TIMESTAMP(timestamp),
+                        RD_KAFKA_V_OPAQUE(msg_opaque),
+                        RD_KAFKA_V_END)
+      );
+}
+
 RdKafka::ErrorCode
 RdKafka::ProducerImpl::produce (const std::string topic_name,
                                 int32_t partition, int msgflags,
                                 void *payload, size_t len,
                                 const void *key, size_t key_len,
                                 int64_t timestamp, void *msg_opaque,
                                 RdKafka::Headers *headers) {
-  rd_kafka_headers_t *hdrs;
+  rd_kafka_headers_t *hdrs = NULL;
   if (headers) {
     hdrs = headers->c_headers();
-  } else {
-    hdrs = 0;
   }
+
   return
     static_cast<RdKafka::ErrorCode>
     (