Merge pull request #12 from ZhuYouzhi/master

Add return code for Kafka::produce and free message payload memory in rd_kafka_example when queue is full
confluentinc · Aug 9, 2013 · bac7e27 · bac7e27
2 parents 52440ea + b6fff10
commit bac7e27
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 5 deletions.
diff --git a/examples/rdkafka_example.c b/examples/rdkafka_example.c
@@ -155,7 +155,11 @@ int main (int argc, char **argv) {
 			strncpy(opbuf, buf, len + 1);
 
 			/* Send/Produce message. */
-			rd_kafka_produce(rk, topic, partition, RD_KAFKA_OP_F_FREE, opbuf, len);
+			if(rd_kafka_produce(rk, topic, partition,
+				RD_KAFKA_OP_F_FREE, opbuf, len) == -1){
+				free(opbuf);
+				continue;
+			}
 			fprintf(stderr, "%% Sent %i bytes to topic "
 				"%s partition %i\n", len, topic, partition);
 			sendcnt++;

diff --git a/examples/rdkafka_example.cpp b/examples/rdkafka_example.cpp
@@ -127,7 +127,11 @@ int main (int argc, char **argv) {
 			strncpy(opbuf, buf, len + 1);
 
 			/* Send/Produce message. */
-			kafka.produce(topic, partition, RD_KAFKA_OP_F_FREE, opbuf, len);
+			if (kafka.produce(topic, partition,
+				RD_KAFKA_OP_F_FREE, opbuf, len) == -1){
+				free(opbuf);
+				continue;
+			}
 			fprintf(stderr, "%% Sent %i bytes to topic "
 				"%s partition %i\n", len, topic, partition);
 			sendcnt++;

diff --git a/rdkafkacpp.h b/rdkafkacpp.h
@@ -30,7 +30,7 @@ class Kafka{
 	const char * getTopic()const{return topic;}
 	*/
 
-	void produce(char *topic, uint32_t partition,int msgflags, char *payload, size_t len);
+	int produce(char *topic, uint32_t partition,int msgflags, char *payload, size_t len);
 
 private:
 	rd_kafka_t *rk;
@@ -56,8 +56,8 @@ bool Kafka::setHandle(rd_kafka_type_t type,const char * broker,const rd_kafka_co
 	return (rk = rd_kafka_new(type, broker, conf))!=NULL;
 }
 
-void Kafka::produce(char *topic, uint32_t partition,int msgflags, char *payload, size_t len){
-	rd_kafka_produce(rk, topic, partition, msgflags, payload, len);
+int Kafka::produce(char *topic, uint32_t partition,int msgflags, char *payload, size_t len){
+	return rd_kafka_produce(rk, topic, partition, msgflags, payload, len);
 }
 
 }