Add support for sending NULL messages (thanks to @secretmike, PR #200)

Author: edenhill

Diff for: src/rdkafka_broker.c

@@ -2127,7 +2127,9 @@ static int rd_kafka_broker_produce_toppar (rd_kafka_broker_t *rkb,
 
 
 		/* Value(payload) length */
-		msghdr->part4.Value_len = htonl(rkm->rkm_len);
+		msghdr->part4.Value_len = htonl(rkm->rkm_payload ?
+                                                rkm->rkm_len :
+                                                RD_KAFKAP_BYTES_LEN_NULL);
 		msghdr->part3.Crc =
 			crc32(msghdr->part3.Crc,
 					(void *)
@@ -2136,14 +2138,16 @@ static int rd_kafka_broker_produce_toppar (rd_kafka_broker_t *rkb,
 					       Value_len));
 
 		rd_kafka_buf_push(rkbuf, &msghdr->part4, sizeof(msghdr->part4));
-			
 
 		/* Payload */
-		msghdr->part3.Crc =
-			crc32(msghdr->part3.Crc,
-					rkm->rkm_payload,
-					rkm->rkm_len);
-		rd_kafka_buf_push(rkbuf, rkm->rkm_payload, rkm->rkm_len);
+                if (rkm->rkm_payload) {
+                        msghdr->part3.Crc =
+                                crc32(msghdr->part3.Crc,
+                                      rkm->rkm_payload,
+                                      rkm->rkm_len);
+                        rd_kafka_buf_push(rkbuf, rkm->rkm_payload,
+                                          rkm->rkm_len);
+                }
 
 
 		/* Finalize Crc */
@@ -2807,7 +2811,10 @@ static rd_kafka_resp_err_t rd_kafka_messageset_handle (rd_kafka_broker_t *rkb,
 					RD_KAFKAP_BYTES_LEN(Key);
 			}
 
-			rko->rko_rkmessage.payload   = Value->data;
+                        /* Forward NULL message notation to application. */
+			rko->rko_rkmessage.payload   =
+                                RD_KAFKAP_BYTES_IS_NULL(Value) ?
+                                NULL : Value->data;
 			rko->rko_rkmessage.len       = Value_len;
 
 			rko->rko_rkmessage.offset    = hdr->Offset;

Diff for: src/rdkafka_msg.c

@@ -40,7 +40,7 @@ void rd_kafka_msg_destroy (rd_kafka_t *rk, rd_kafka_msg_t *rkm) {
 	rd_kafka_assert(rk, rk->rk_producer.msg_cnt > 0);
 	(void)rd_atomic_sub(&rk->rk_producer.msg_cnt, 1);
 
-	if (rkm->rkm_flags & RD_KAFKA_MSG_F_FREE)
+	if (rkm->rkm_flags & RD_KAFKA_MSG_F_FREE && rkm->rkm_payload)
 		free(rkm->rkm_payload);
 
 	if (rkm->rkm_key)
@@ -93,7 +93,7 @@ static rd_kafka_msg_t *rd_kafka_msg_new0 (rd_kafka_topic_t *rkt,
 			rkt->rkt_conf.message_timeout_ms * 1000;
 	}
 
-	if (msgflags & RD_KAFKA_MSG_F_COPY) {
+	if (payload && msgflags & RD_KAFKA_MSG_F_COPY) {
 		/* Copy payload to space following the ..msg_t */
 		rkm->rkm_payload = (void *)(rkm+1);
 		memcpy(rkm->rkm_payload, payload, len);