Let produce API honour "message.max.bytes" (issue #24)

Producing a message larger than "message.max.bytes" will now make
rd_kafka_produce() return -1 and set errno to EMSGSIZE.

examples/rdkafka_performance.c

@@ -568,6 +568,10 @@ int main (int argc, char **argv) {
 					       errno == ENOBUFS ?
 					       " (backpressure)":"");
 				cnt.tx_err++;
+				if (errno != ENOBUFS) {
+					run = 0;
+					break;
+				}
 				now = rd_clock();
 				if (cnt.t_last + dispintvl <= now) {
 					printf("%% Backpressure %i "

rdkafka.c

@@ -842,17 +842,6 @@ rd_kafka_t *rd_kafka_new (rd_kafka_type_t type, rd_kafka_conf_t *conf,
 
 /**
  * Produce a single message.
- *
- * If 'partition' is unassigned (RD_KAFKA_PARTITION_UA) the configured or
- * default partitioner will be used to designate the target partition.
- *
- * See rdkafka.h for 'msgflags'.
- *
- * Returns: 0 on success or -1 on error (see errno for details)
- *
- * errnos:
- *    ENOBUFS - conf.producer.max_msg_cnt would be exceeded.
- *
  * Locality: any application thread
  */
 int rd_kafka_produce (rd_kafka_topic_t *rkt, int32_t partition,

rdkafka.h

@@ -634,8 +634,11 @@ int rd_kafka_consume_callback (rd_kafka_topic_t *rkt, int32_t partition,
  * pointer that will provided in the delivery report callback (`dr_cb`) for
  * referencing this message.
  *
- * Returns 0 on success or -1 if the maximum number of outstanding messages
- * (conf.producer.max_messages) has been reached (`errno==ENOBUFS`).
+ * Returns 0 on success or -1 on error in which case errno is set accordingly:
+ *   ENOBUFS  - maximum number of outstanding messages has been reached:
+ *              "queue.buffering.max.message"
+ *   EMSGSIZE - message is larger than configured max size:
+ *              "messages.max.bytes".
  *
  */
 

rdkafka_broker.c

@@ -1732,6 +1732,12 @@ static int rd_kafka_broker_produce_toppar (rd_kafka_broker_t *rkb,
 		msghdr++;
 	}
 
+	/* No messages added, bail out early. */
+	if (unlikely(rkbuf->rkbuf_msgq.rkmq_msg_cnt == 0)) {
+		rd_kafka_buf_destroy(rkbuf);
+		return -1;
+	}
+
 	/* Compress the messages */
 	if (rkb->rkb_rk->rk_conf.compression_codec) {
 		int    siovlen = 1;
@@ -2104,8 +2110,10 @@ static void rd_kafka_broker_producer_serve (rd_kafka_broker_t *rkb) {
 				while (rktp->rktp_xmit_msgq.rkmq_msg_cnt > 0) {
 					int r = rd_kafka_broker_produce_toppar(
 						rkb, rktp);
-					if (r > 0)
+					if (likely(r > 0))
 						cnt += r;
+					else
+						break;
 				}
 			}
 

rdkafka_msg.c

@@ -71,6 +71,11 @@ int rd_kafka_msg_new (rd_kafka_topic_t *rkt, int32_t force_partition,
 		return -1;
 	}
 
+	if (unlikely(len + keylen > rkt->rkt_rk->rk_conf.max_msg_size)) {
+		errno = EMSGSIZE;
+		return -1;
+	}
+
 	/* If we are to make a copy of the payload, allocate space for it too */
 	if (msgflags & RD_KAFKA_MSG_F_COPY) {
 		msgflags &= ~RD_KAFKA_MSG_F_FREE;