Merge lp:~noskcaj/ubuntu/trusty/librdkafka/0.8.1 into lp:ubuntu/trusty/librdkafka

[Jackson Doak]

New upstream release, should fix ftbfs. I cannot test as i don't have working PPC hardware, but http://bugs.debian.org/cgi-bin/bugreport.cgi?bug=730506 suggests this should work.

[Martin Pitt (pitti)]

Thanks!

review: Approve

=== modified file 'CONFIGURATION.md'
--- CONFIGURATION.md	2013-11-04 16:50:07 +0000
+++ CONFIGURATION.md	2014-01-01 03:04:53 +0000
@@ -11,6 +11,8 @@
 topic.metadata.refresh.fast.interval.ms  |           250 | See `topic.metadata.refresh.fast.cnt` description
 debug                                    |               | A comma-separated list of debug contexts to enable: all,generic,broker,topic,metadata,producer,queue,msg
 socket.timeout.ms                        |         60000 | Timeout for network requests.
+socket.send.buffer.bytes                 |             0 | Broker socket send buffer size. System default is used if 0.
+socket.receive.buffer.bytes              |             0 | Broker socket receive buffer size. System default is used if 0.
 broker.address.ttl                       |        300000 | How long to cache the broker address resolving results.
 statistics.interval.ms                   |             0 | librdkafka statistics emit interval. The application also needs to register a stats callback using `rd_kafka_conf_set_stats_cb()`. The granularity is 1000ms.
 queued.min.messages                      |        100000 | Minimum number of messages that should to be available for consumption by application.

=== modified file 'Makefile'
--- Makefile	2013-11-04 16:50:07 +0000
+++ Makefile	2014-01-01 03:04:53 +0000
@@ -17,7 +17,9 @@
 CFLAGS+=-g
 
 # Clang warnings to ignore
-CFLAGS+=-Wno-gnu-designator
+ifeq ($(CC),clang)
+	CFLAGS+=-Wno-gnu-designator
+endif
 
 # Enable iovecs in snappy
 CFLAGS+=-DSG
@@ -82,5 +84,6 @@
 	rm -f $(OBJS) $(DEPS) \
 		$(LIBNAME)*.a $(LIBNAME)*.so $(LIBNAME)*.so.$(LIBVER)
 	make -C tests clean
+	make -C examples clean
 
 -include $(DEPS)

=== modified file 'README.md'
--- README.md	2013-11-04 16:50:07 +0000
+++ README.md	2014-01-01 03:04:53 +0000
@@ -24,12 +24,12 @@
   * Producer: supported
   * Consumer: supported
   * Compression: snappy and gzip
-  * Debian package: work in progress (separate "debian" branch)
+  * Debian package: "debian" branch ([ITP #710271](http://bugs.debian.org/cgi-bin/bugreport.cgi?bug=710271))
   * ZooKeeper: not supported
-  * API: not backwards compatible
+  * API: Stable, not backwards compatible
   * Tests: Regression tests in `tests/` directory.
   * Statistics: JSON formatted, see `rd_kafka_conf_set_stats_cb` in `rdkafka.h`.
-  * Status: Testing, APIs subject to change.
+  * Status: Stable
 
 
 **Apache Kafka 0.7 support:**
@@ -117,10 +117,10 @@
 [GitHub Issues](https://github.com/edenhill/librdkafka/issues)
 
 
-Questions and discussions are also welcome on irc.freenode.org, #kafka,
+Questions and discussions are also welcome on irc.freenode.org, #apache-kafka,
 nickname Snaps.
 
 
-### Commerical support
+### Commercial support
 
 Commercial support is available from [Edenhill services](http://www.edenhill.se)

=== modified file 'debian/changelog'
--- debian/changelog	2013-11-04 16:50:07 +0000
+++ debian/changelog	2014-01-01 03:04:53 +0000
@@ -1,3 +1,11 @@
+librdkafka (0.8.1-0ubuntu1) trusty; urgency=medium
+
+  * New upstream release.
+    - Should fix FTBFS at debian bug 730506
+  * Update symbols
+
+ -- Jackson Doak <[e-mail]>  Wed, 01 Jan 2014 13:55:40 +1100
+
 librdkafka (0.8.0-1) unstable; urgency=low
 
   * Initial release. (Closes: #710271)

=== modified file 'debian/control'
--- debian/control	2013-11-04 16:50:07 +0000
+++ debian/control	2014-01-01 03:04:53 +0000
@@ -1,6 +1,7 @@
 Source: librdkafka
 Priority: optional
-Maintainer: Faidon Liambotis <[e-mail]>
+Maintainer: Ubuntu Developers <[e-mail]>
+XSBC-Original-Maintainer: Faidon Liambotis <[e-mail]>
 Build-Depends: debhelper (>= 9), zlib1g-dev
 Standards-Version: 3.9.4
 Section: libs

=== modified file 'debian/librdkafka1.symbols'
--- debian/librdkafka1.symbols	2013-11-04 16:50:07 +0000
+++ debian/librdkafka1.symbols	2014-01-01 03:04:53 +0000
@@ -84,4 +84,6 @@
  rd_kafka_toppar_insert_msg@Base 0.8.0
  rd_kafka_toppar_insert_msgq@Base 0.8.0
  rd_kafka_toppar_ua_move@Base 0.8.0
+ rd_kafka_version@Base 0.8.1
+ rd_kafka_version_str@Base 0.8.1
  rd_kafka_wait_destroyed@Base 0.8.0

=== modified file 'examples/rdkafka_example.c'
--- examples/rdkafka_example.c	2013-11-04 16:50:07 +0000
+++ examples/rdkafka_example.c	2014-01-01 03:04:53 +0000
@@ -214,6 +214,8 @@
 			"Usage: %s [-C|-P] -t <topic> "
 			"[-p <partition>] [-b <host1:port1,host2:port2,..>]\n"
 			"\n"
+			"librdkafka version %s (0x%08x)\n"
+			"\n"
 			" Options:\n"
 			"  -C | -P         Consumer or Producer mode\n"
 			"  -t <topic>      Topic to fetch / produce\n"
@@ -231,8 +233,11 @@
 			"  writes fetched messages to stdout\n"
 			" In Producer mode:\n"
 			"  reads messages from stdin and sends to broker\n"
+			"\n"
+			"\n"
 			"\n",
 			argv[0],
+			rd_kafka_version_str(), rd_kafka_version(),
 			RD_KAFKA_DEBUG_CONTEXTS);
 		exit(1);
 	}

=== modified file 'examples/rdkafka_performance.c'
--- examples/rdkafka_performance.c	2013-11-04 16:50:07 +0000
+++ examples/rdkafka_performance.c	2014-01-01 03:04:53 +0000
@@ -87,7 +87,6 @@
 			   void *payload, size_t len,
 			   int error_code,
 			   void *opaque, void *msg_opaque) {
-	long int msgid = (long int)msg_opaque;
 	static rd_ts_t last;
 	rd_ts_t now = rd_clock();
 	static int msgs;
@@ -106,12 +105,12 @@
 	    !(msgs_wait_cnt % (dispintvl / 1000)) || 
 	    (now - last) >= dispintvl * 1000) {
 		if (error_code)
-			printf("Message %ld delivey failed: %s (%li remain)\n",
-			       msgid, rd_kafka_err2str(error_code),
+			printf("Message delivey failed: %s (%li remain)\n",
+			       rd_kafka_err2str(error_code),
 			       msgs_wait_cnt);
 		else if (!quiet)
-			printf("Message %ld delivered: %li remain\n",
-			       msgid, msgs_wait_cnt);
+			printf("Message delivered: %li remain\n",
+			       msgs_wait_cnt);
 		if (!quiet && do_seq)
 			printf(" --> \"%.*s\"\n", (int)len, (char *)payload);
 		last = now;
@@ -413,6 +412,8 @@
 			"Usage: %s [-C|-P] -t <topic> "
 			"[-p <partition>] [-b <broker,broker..>] [options..]\n"
 			"\n"
+			"librdkafka version %s (0x%08x)\n"
+			"\n"
 			" Options:\n"
 			"  -C | -P      Consumer or Producer mode\n"
 			"  -t <topic>   Topic to fetch / produce\n"
@@ -454,6 +455,7 @@
 			"  writes messages of size -s <..> and prints thruput\n"
 			"\n",
 			argv[0],
+			rd_kafka_version_str(), rd_kafka_version(),
 			RD_KAFKA_DEBUG_CONTEXTS);
 		exit(1);
 	}
@@ -560,14 +562,17 @@
 			while (run &&
 			       rd_kafka_produce(rkt, partition,
 						sendflags, pbuf, msgsize,
-						key, keylen,
-						(void *)cnt.msgs) == -1) {
+						key, keylen, NULL) == -1) {
 				if (!quiet || errno != ENOBUFS)
 					printf("produce error: %s%s\n",
 					       strerror(errno),
 					       errno == ENOBUFS ?
 					       " (backpressure)":"");
 				cnt.tx_err++;
+				if (errno != ENOBUFS) {
+					run = 0;
+					break;
+				}
 				now = rd_clock();
 				if (cnt.t_last + dispintvl <= now) {
 					printf("%% Backpressure %i "
@@ -633,7 +638,7 @@
 		 * Consumer
 		 */
 
-		rd_kafka_message_t **rkmessages;
+		rd_kafka_message_t **rkmessages = NULL;
 
 #if 0 /* Future API */
 		/* The offset storage file is optional but its presence
@@ -725,6 +730,9 @@
 					strerror(errno));
 
 			print_stats(mode, 0, compression);
+
+			/* Poll to handle stats callbacks */
+			rd_kafka_poll(rk, 0);
 		}
 		cnt.t_end = rd_clock();
 

=== modified file 'rdkafka.c'
--- rdkafka.c	2013-11-04 16:50:07 +0000
+++ rdkafka.c	2014-01-01 03:04:53 +0000
@@ -353,7 +353,7 @@
 	rd_kafka_dbg(rk, QUEUE, "QSERVE", "Serving %i ops", localq.rkq_qlen);
 
 	/* Call callback for each op */
-	TAILQ_FOREACH_SAFE(rko, tmp, &localq.rkq_q, rko_link) {
+	TAILQ_FOREACH_SAFE(rko, &localq.rkq_q, rko_link, tmp) {
 		callback(rko, opaque);
 		rd_kafka_op_destroy(rko);
 	}
@@ -515,7 +515,7 @@
 
 	/* Decommission all topics */
 	rd_kafka_lock(rk);
-	TAILQ_FOREACH_SAFE(rkt, rkt_tmp, &rk->rk_topics, rkt_link) {
+	TAILQ_FOREACH_SAFE(rkt, &rk->rk_topics, rkt_link, rkt_tmp) {
 		rd_kafka_unlock(rk);
 		rd_kafka_topic_partitions_remove(rkt);
 		rd_kafka_lock(rk);
@@ -553,7 +553,7 @@
 	size_t size = *sizep;
 	int of = *ofp;
 
-	_st_printf("%s{ "
+	_st_printf("%s\"%"PRId32"\": { "
 		   "\"partition\":%"PRId32", "
 		   "\"leader\":%"PRId32", "
 		   "\"desired\":%s, "
@@ -574,6 +574,7 @@
 		   "} ",
 		   first ? "" : ", ",
 		   rktp->rktp_partition,
+		   rktp->rktp_partition,
 		   rktp->rktp_leader ? rktp->rktp_leader->rkb_nodeid : -1,
 		   (rktp->rktp_flags&RD_KAFKA_TOPPAR_F_DESIRED)?"true":"false",
 		   (rktp->rktp_flags&RD_KAFKA_TOPPAR_F_UNKNOWN)?"true":"false",
@@ -618,7 +619,7 @@
 		   "\"ts\":%"PRIu64", "
 		   "\"time\":%lli, "
 		   "\"replyq\":%i, "
-		   "\"brokers\":[ "/*open brokers*/,
+		   "\"brokers\":{ "/*open brokers*/,
 		   now,
 		   (signed long long)time(NULL),
 		   rk->rk_rep.rkq_qlen);
@@ -626,7 +627,8 @@
 
 	TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {
 		rd_kafka_broker_lock(rkb);
-		_st_printf("%s{ "/*open broker*/
+		rd_kafka_avg_rollover(&rkb->rkb_rtt_last, &rkb->rkb_rtt_curr);
+		_st_printf("%s\"%s\": { "/*open broker*/
 			   "\"name\":\"%s\", "
 			   "\"nodeid\":%"PRId32", "
 			   "\"state\":\"%s\", "
@@ -639,9 +641,16 @@
 			   "\"rx\":%"PRIu64", "
 			   "\"rxbytes\":%"PRIu64", "
 			   "\"rxerrs\":%"PRIu64", "
-			   "\"toppars\":[ "/*open toppars*/,
+			   "\"rtt\": {"
+			   " \"min\":%"PRIu64","
+			   " \"max\":%"PRIu64","
+			   " \"avg\":%"PRIu64","
+			   " \"cnt\":%i "
+			   "}, "
+			   "\"toppars\":{ "/*open toppars*/,
 			   rkb == TAILQ_FIRST(&rk->rk_brokers) ? "" : ", ",
 			   rkb->rkb_name,
+			   rkb->rkb_name,
 			   rkb->rkb_nodeid,
 			   rd_kafka_broker_state_names[rkb->rkb_state],
 			   rkb->rkb_outbufs.rkbq_cnt,
@@ -652,37 +661,43 @@
 			   rkb->rkb_c.tx_retries,
 			   rkb->rkb_c.rx,
 			   rkb->rkb_c.rx_bytes,
-			   rkb->rkb_c.rx_err);
+			   rkb->rkb_c.rx_err,
+			   rkb->rkb_rtt_last.ra_min,
+			   rkb->rkb_rtt_last.ra_max,
+			   rkb->rkb_rtt_last.ra_avg,
+			   rkb->rkb_rtt_last.ra_cnt);
 
 		rd_kafka_broker_toppars_rdlock(rkb);
 		TAILQ_FOREACH(rktp, &rkb->rkb_toppars, rktp_rkblink) {
-			_st_printf("%s{ "
+			_st_printf("%s\"%.*s\": { "
 				   "\"topic\":\"%.*s\", "
 				   "\"partition\":%"PRId32"} ",
 				   rktp==TAILQ_FIRST(&rkb->rkb_toppars)?"":", ",
 				   RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
+				   RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
 				   rktp->rktp_partition);
 		}
 		rd_kafka_broker_toppars_unlock(rkb);
 
 		rd_kafka_broker_unlock(rkb);
 
-		_st_printf("] "/*close toppars*/
+		_st_printf("} "/*close toppars*/
 			   "} "/*close broker*/);
 	}
 
 
-	_st_printf("], " /* close "brokers" array */
-		   "\"topics\":[ ");
+	_st_printf("}, " /* close "brokers" array */
+		   "\"topics\":{ ");
 
 	TAILQ_FOREACH(rkt, &rk->rk_topics, rkt_link) {
 		int i;
 
 		rd_kafka_topic_rdlock(rkt);
-		_st_printf("%s{ "
+		_st_printf("%s\"%.*s\": { "
 			   "\"topic\":\"%.*s\", "
-			   "\"partitions\":[ " /*open partitions*/,
+			   "\"partitions\":{ " /*open partitions*/,
 			   rkt==TAILQ_FIRST(&rk->rk_topics)?"":", ",
+			   RD_KAFKAP_STR_PR(rkt->rkt_topic),
 			   RD_KAFKAP_STR_PR(rkt->rkt_topic));
 
 		for (i = 0 ; i < rkt->rkt_partition_cnt ; i++)
@@ -699,14 +714,14 @@
 						   rkt->rkt_ua, i++ == 0);
 		rd_kafka_topic_unlock(rkt);
 
-		_st_printf("] "/*close partitions*/
+		_st_printf("} "/*close partitions*/
 			   "} "/*close topic*/);
 
 	}
 
 	rd_kafka_unlock(rk);
 
-	_st_printf("] "/*close topics*/
+	_st_printf("} "/*close topics*/
 		   "}"/*close object*/);
 
 
@@ -842,17 +857,6 @@
 
 /**
  * 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,
@@ -1279,3 +1283,21 @@
 int rd_kafka_outq_len (rd_kafka_t *rk) {
 	return rk->rk_producer.msg_cnt;
 }
+
+
+int rd_kafka_version (void) {
+	return RD_KAFKA_VERSION;
+}
+
+const char *rd_kafka_version_str (void) {
+	static char ret[64];
+	int ver = rd_kafka_version();
+
+	if (!*ret)
+		snprintf(ret, sizeof(ret), "%i.%i.%i",
+			 (ver >> 24) & 0xff,
+			 (ver >> 16) & 0xff,
+			 (ver >> 8) & 0xff);
+
+	return ret;
+}

=== modified file 'rdkafka.h'
--- rdkafka.h	2013-11-04 16:50:07 +0000
+++ rdkafka.h	2014-01-01 03:04:53 +0000
@@ -43,13 +43,25 @@
 /**
  * librdkafka version
  *
- * interpreted as MM.mm.rr.xx:
+ * Interpreted as hex MM.mm.rr.xx:
  *   MM = Major
  *   mm = minor
  *   rr = revision
  *   xx = currently unused
- */
-#define RD_KAFKA_VERSION  0x00080000
+ *
+ * I.e.: 0x00080100 = 0.8.1
+ */
+#define RD_KAFKA_VERSION  0x00080100
+
+/**
+ * Returns the librdkafka version as integer.
+ */
+int rd_kafka_version (void);
+
+/**
+ * Returns the librdkafka version as string.
+ */
+const char *rd_kafka_version_str (void);
 
 
 /**
@@ -634,8 +646,11 @@
  * 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".
  *
  */
 

=== modified file 'rdkafka_broker.c'
--- rdkafka_broker.c	2013-11-04 16:50:07 +0000
+++ rdkafka_broker.c	2014-01-01 03:04:53 +0000
@@ -72,7 +72,7 @@
 	int i;
 
 	rd_kafka_dbg(rk, MSG, "MSG", "%s: iovlen %zd",
-		     what, msg->msg_iovlen);
+		     what, (size_t)msg->msg_iovlen);
 
 	for (i = 0 ; i < msg->msg_iovlen ; i++) {
 		rd_kafka_dbg(rk, MSG, what,
@@ -221,26 +221,32 @@
 	TAILQ_INIT(&rkbufq->rkbq_bufs);
 	rkbufq->rkbq_cnt = 0;
 }
+
+/**
+ * Concat all buffers from 'src' to tail of 'dst'
+ */
+static void rd_kafka_bufq_concat (rd_kafka_bufq_t *dst, rd_kafka_bufq_t *src) {
+	TAILQ_CONCAT(&dst->rkbq_bufs, &src->rkbq_bufs, rkbuf_link);
+	rd_atomic_add(&dst->rkbq_cnt, src->rkbq_cnt);
+	rd_kafka_bufq_init(src);
+}
+
 /**
  * Purge the wait-response queue.
+ * NOTE: 'rkbufq' must be a temporary queue and not one of rkb_waitresps
+ *       or rkb_outbufs since buffers may be re-enqueued on those queues.
  */
 static void rd_kafka_bufq_purge (rd_kafka_broker_t *rkb,
 				 rd_kafka_bufq_t *rkbufq,
 				 rd_kafka_resp_err_t err) {
 	rd_kafka_buf_t *rkbuf, *tmp;
-	rd_kafka_bufq_t tmpq;
 
 	assert(pthread_self() == rkb->rkb_thread);
 
-	/* Move messages to temporary queue to avoid infinite loops if they
-	 * are requeued on the same queue in the rkbuf_cb callback. */
-	rd_kafka_bufq_init(&tmpq);
-	TAILQ_MOVE(&tmpq.rkbq_bufs, &rkbufq->rkbq_bufs, rkbuf_link);
-	rd_kafka_bufq_init(rkbufq);
-
-	rd_rkb_dbg(rkb, QUEUE, "BUFQ", "Purging bufq");
-
-	TAILQ_FOREACH_SAFE(rkbuf, tmp, &tmpq.rkbq_bufs, rkbuf_link)
+	rd_rkb_dbg(rkb, QUEUE, "BUFQ", "Purging bufq with %i buffers",
+		   rkbufq->rkbq_cnt);
+
+	TAILQ_FOREACH_SAFE(rkbuf, &rkbufq->rkbq_bufs, rkbuf_link, tmp)
 		rkbuf->rkbuf_cb(rkb, err, NULL, rkbuf, rkbuf->rkbuf_opaque);
 }
 
@@ -255,14 +261,14 @@
 
 	assert(pthread_self() == rkb->rkb_thread);
 
-	TAILQ_FOREACH_SAFE(rkbuf, tmp,
-			   &rkb->rkb_waitresps.rkbq_bufs, rkbuf_link) {
+	TAILQ_FOREACH_SAFE(rkbuf,
+			   &rkb->rkb_waitresps.rkbq_bufs, rkbuf_link, tmp) {
 		if (likely(rkbuf->rkbuf_ts_timeout > now))
 			continue;
 
 		rd_kafka_bufq_deq(&rkb->rkb_waitresps, rkbuf);
 
-		rkbuf->rkbuf_cb(rkb, RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT,
+		rkbuf->rkbuf_cb(rkb, RD_KAFKA_RESP_ERR__MSG_TIMED_OUT,
 				NULL, rkbuf, rkbuf->rkbuf_opaque);
 		cnt++;
 	}
@@ -286,6 +292,7 @@
 	va_list ap;
 	int errno_save = errno;
 	rd_kafka_toppar_t *rktp;
+	rd_kafka_bufq_t tmpq;
 
 	assert(pthread_self() == rkb->rkb_thread);
 	rd_kafka_broker_lock(rkb);
@@ -338,11 +345,21 @@
 				  strlen(rkb->rkb_err.msg));
 	}
 
-	rd_kafka_bufq_purge(rkb, &rkb->rkb_waitresps, err);
-	rd_kafka_bufq_purge(rkb, &rkb->rkb_outbufs, err);
+	/*
+	 * Purge all buffers
+	 * (put on a temporary queue since bufs may be requeued)
+	 */
+	rd_kafka_bufq_init(&tmpq);
+	rd_kafka_bufq_concat(&tmpq, &rkb->rkb_waitresps);
+	rd_kafka_bufq_concat(&tmpq, &rkb->rkb_outbufs);
 
+	/* Unlock broker since a requeue will try to lock it. */
 	rd_kafka_broker_unlock(rkb);
 
+	/* Purge the buffers */
+	rd_kafka_bufq_purge(rkb, &tmpq, err);
+
+
 	/* Undelegate all toppars from this broker. */
 	rd_kafka_broker_toppars_wrlock(rkb);
 	while ((rktp = TAILQ_FIRST(&rkb->rkb_toppars))) {
@@ -382,7 +399,7 @@
 
 		rd_kafka_dbg(rkb->rkb_rk, BROKER, "BRKSEND",
 			     "sendmsg FAILED for iovlen %zd (%i)",
-			     msg->msg_iovlen,
+			     (size_t)msg->msg_iovlen,
 			     IOV_MAX);
 		rd_kafka_broker_fail(rkb, RD_KAFKA_RESP_ERR__TRANSPORT,
 				     "Send failed: %s", strerror(errno));
@@ -390,8 +407,8 @@
 		return -1;
 	}
 
-	rkb->rkb_c.tx_bytes += r;
-	rkb->rkb_c.tx++;
+	rd_atomic_add(&rkb->rkb_c.tx_bytes, r);
+	rd_atomic_add(&rkb->rkb_c.tx, 1);
 	return r;
 }
 
@@ -1020,11 +1037,15 @@
 static rd_kafka_buf_t *rd_kafka_waitresp_find (rd_kafka_broker_t *rkb,
 					       int32_t corrid) {
 	rd_kafka_buf_t *rkbuf;
+	rd_ts_t now = rd_clock();
 
 	assert(pthread_self() == rkb->rkb_thread);
 
 	TAILQ_FOREACH(rkbuf, &rkb->rkb_waitresps.rkbq_bufs, rkbuf_link)
 		if (rkbuf->rkbuf_corrid == corrid) {
+			rd_kafka_avg_add(&rkb->rkb_rtt_curr,
+					 now - rkbuf->rkbuf_ts_sent);
+
 			rd_kafka_bufq_deq(&rkb->rkb_waitresps, rkbuf);
 			return rkbuf;
 		}
@@ -1089,10 +1110,13 @@
 		off_t vof = of - len;
 
 		if (0)
-			printf(" #%i/%zd and %zd: of %zd, len %zd, "
-			       "vof %zd: iov %zd\n",
-			       i, src->msg_iovlen, dst->msg_iovlen,
-			       of, len, vof, src->msg_iov[i].iov_len);
+			printf(" #%i/%zd and %zd: of %jd, len %zd, "
+			       "vof %jd: iov %zd\n",
+			       i,
+			       (size_t)src->msg_iovlen,
+			       (size_t)dst->msg_iovlen,
+			       (intmax_t)of, len, (intmax_t)vof,
+			       src->msg_iov[i].iov_len);
 		if (vof < 0)
 			vof = 0;
 
@@ -1248,8 +1272,8 @@
 	if (rkbuf->rkbuf_of == rkbuf->rkbuf_len + sizeof(rkbuf->rkbuf_reshdr)) {
 		/* Message is complete, pass it on to the original requester. */
 		rkb->rkb_recv_buf = NULL;
-		rkb->rkb_c.rx++;
-		rkb->rkb_c.rx_bytes += rkbuf->rkbuf_of;
+		rd_atomic_add(&rkb->rkb_c.rx, 1);
+		rd_atomic_add(&rkb->rkb_c.rx_bytes, rkbuf->rkbuf_of);
 		rd_kafka_req_response(rkb, rkbuf);
 	}
 
@@ -1314,6 +1338,32 @@
 	rd_rkb_dbg(rkb, BROKER, "CONNECTED", "connected to %s",
 		   rd_sockaddr2str(sinx, RD_SOCKADDR2STR_F_NICE));
 
+	/* Set socket send & receive buffer sizes if configuerd */
+	if (rkb->rkb_rk->rk_conf.socket_sndbuf_size != 0) {
+		if (setsockopt(rkb->rkb_s, SOL_SOCKET, SO_SNDBUF,
+			       &rkb->rkb_rk->rk_conf.socket_sndbuf_size,
+			       sizeof(rkb->rkb_rk->rk_conf.
+				      socket_sndbuf_size)) == -1)
+			rd_rkb_log(rkb, LOG_WARNING, "SNDBUF",
+				   "Failed to set socket send "
+				   "buffer size to %i: %s",
+				   rkb->rkb_rk->rk_conf.socket_sndbuf_size,
+				   strerror(errno));
+	}
+
+	if (rkb->rkb_rk->rk_conf.socket_rcvbuf_size != 0) {
+		if (setsockopt(rkb->rkb_s, SOL_SOCKET, SO_RCVBUF,
+			       &rkb->rkb_rk->rk_conf.socket_rcvbuf_size,
+			       sizeof(rkb->rkb_rk->rk_conf.
+				      socket_rcvbuf_size)) == -1)
+			rd_rkb_log(rkb, LOG_WARNING, "RCVBUF",
+				   "Failed to set socket receive "
+				   "buffer size to %i: %s",
+				   rkb->rkb_rk->rk_conf.socket_rcvbuf_size,
+				   strerror(errno));
+	}
+
+
 	rd_kafka_broker_lock(rkb);
 	rd_kafka_broker_set_state(rkb, RD_KAFKA_BROKER_STATE_UP);
 	rkb->rkb_err.err = 0;
@@ -1381,6 +1431,9 @@
 		/* Entire buffer sent, unlink from outbuf */
 		rd_kafka_bufq_deq(&rkb->rkb_outbufs, rkbuf);
 
+		/* Store time for RTT calculation */
+		rkbuf->rkbuf_ts_sent = rd_clock();
+
 		/* Put buffer on response wait list unless we are not
 		 * expecting a response (required_acks=0). */
 		if (!(rkbuf->rkbuf_flags & RD_KAFKA_OP_F_NO_RESPONSE))
@@ -1606,7 +1659,7 @@
 	if (0)
 		rd_rkb_dbg(rkb, MSG, "PRODUCE",
 			   "Serve %i/%i messages (%i iovecs) "
-			   "for %.*s [%"PRId32"] (%zd bytes)",
+			   "for %.*s [%"PRId32"] (%"PRIu64" bytes)",
 			   msgcnt, rktp->rktp_msgq.rkmq_msg_cnt,
 			   iovcnt,
 			   RD_KAFKAP_STR_PR(rkt->rkt_topic),
@@ -1732,6 +1785,12 @@
 		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;
@@ -1923,8 +1982,8 @@
 
 do_send:
 
-	rktp->rktp_c.tx_msgs  += rkbuf->rkbuf_msgq.rkmq_msg_cnt;
-	rktp->rktp_c.tx_bytes += prodhdr->part2.MessageSetSize;
+	rd_atomic_add(&rktp->rktp_c.tx_msgs, rkbuf->rkbuf_msgq.rkmq_msg_cnt);
+	rd_atomic_add(&rktp->rktp_c.tx_bytes, prodhdr->part2.MessageSetSize);
 
 	prodhdr->part2.MessageSetSize =
 		htonl(prodhdr->part2.MessageSetSize);
@@ -2104,8 +2163,10 @@
 				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;
 				}
 			}
 
@@ -2114,7 +2175,7 @@
 		/* Trigger delivery report for timed out messages */
 		if (unlikely(timedout.rkmq_msg_cnt > 0))
 			rd_kafka_dr_msgq(rkb->rkb_rk, &timedout,
-					 RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT);
+					 RD_KAFKA_RESP_ERR__MSG_TIMED_OUT);
 
 		rd_kafka_broker_toppars_unlock(rkb);
 
@@ -3169,10 +3230,11 @@
 
 int rd_kafka_brokers_add (rd_kafka_t *rk, const char *brokerlist) {
 	char *s = strdupa(brokerlist);
-	char *t, *n;
+	char *t, *t2, *n;
 	int cnt = 0;
 	rd_kafka_broker_t *rkb;
 
+	/* Parse comma-separated list of brokers. */
 	while (*s) {
 		uint16_t port = 0;
 
@@ -3186,7 +3248,17 @@
 		else
 			n = s + strlen(s)-1;
 
-		if ((t = strchr(s, ':'))) {
+		/* Check if port has been specified, but try to identify IPv6
+		 * addresses first:
+		 *  t = last ':' in string
+		 *  t2 = first ':' in string
+		 *  If t and t2 are equal then only one ":" exists in name
+		 *  and thus an IPv4 address with port specified.
+		 *  Else if not equal and t is prefixed with "]" then it's an
+		 *  IPv6 address with port specified.
+		 *  Else no port specified. */
+		if ((t = strrchr(s, ':')) &&
+		    ((t2 = strchr(s, ':')) == t || *(t-1) == ']')) {
 			*t = '\0';
 			port = atoi(t+1);
 		}

=== modified file 'rdkafka_defaultconf.c'
--- rdkafka_defaultconf.c	2013-11-04 16:50:07 +0000
+++ rdkafka_defaultconf.c	2014-01-01 03:04:53 +0000
@@ -122,6 +122,14 @@
 	{ _RK_GLOBAL, "socket.timeout.ms", _RK_C_INT, _RK(socket_timeout_ms),
 	  "Timeout for network requests.",
 	  10, 300*1000, 60*1000 },
+	{ _RK_GLOBAL, "socket.send.buffer.bytes", _RK_C_INT,
+	  _RK(socket_sndbuf_size),
+	  "Broker socket send buffer size. System default is used if 0.",
+	  0, 100000000, 0 },
+	{ _RK_GLOBAL, "socket.receive.buffer.bytes", _RK_C_INT,
+	  _RK(socket_rcvbuf_size),
+	  "Broker socket receive buffer size. System default is used if 0.",
+	  0, 100000000, 0 },
 	{ _RK_GLOBAL, "broker.address.ttl", _RK_C_INT,
 	  _RK(broker_addr_ttl),
 	  "How long to cache the broker address resolving results.",

=== modified file 'rdkafka_int.h'
--- rdkafka_int.h	2013-11-04 16:50:07 +0000
+++ rdkafka_int.h	2014-01-01 03:04:53 +0000
@@ -123,6 +123,8 @@
 	int     debug;
 	int     broker_addr_ttl;
 	int     socket_timeout_ms;
+	int     socket_sndbuf_size;
+	int     socket_rcvbuf_size;
 	char   *clientid;
 	char   *brokerlist;
 	int     stats_interval_ms;
@@ -205,6 +207,43 @@
 
 
 
+typedef struct rd_kafka_avg_s {
+	rd_ts_t ra_max;
+	rd_ts_t ra_min;
+	rd_ts_t ra_avg;
+	rd_ts_t ra_sum;
+	int     ra_cnt;
+} rd_kafka_avg_t;
+
+/**
+ * Add timestamp 'ts' to averager 'ra'.
+ */
+static RD_UNUSED void rd_kafka_avg_add (rd_kafka_avg_t *ra, rd_ts_t ts) {
+	if (ts > ra->ra_max)
+		ra->ra_max = ts;
+	if (ra->ra_min == 0 || ts < ra->ra_min)
+		ra->ra_min = ts;
+	ra->ra_sum += ts;
+	ra->ra_cnt++;
+}
+
+/**
+ * Rolls over statistics in 'src' and stores the average in 'dst'.
+ * 'src' is cleared and ready to be reused.
+ */
+static RD_UNUSED void rd_kafka_avg_rollover (rd_kafka_avg_t *dst,
+					     rd_kafka_avg_t *src) {
+	*dst = *src;
+	if (dst->ra_cnt)
+		dst->ra_avg = dst->ra_sum / dst->ra_cnt;
+	else
+		dst->ra_avg = 0;
+
+	memset(src, 0, sizeof(*src));
+}
+
+
+
 typedef struct rd_kafka_msg_s {
 	TAILQ_ENTRY(rd_kafka_msg_s)  rkm_link;
 	int        rkm_flags;
@@ -405,6 +444,9 @@
 	rd_kafka_bufq_t     rkb_waitresps;
 	rd_kafka_bufq_t     rkb_retrybufs;
 
+	rd_kafka_avg_t      rkb_rtt_curr;       /* Current averaging period */
+	rd_kafka_avg_t      rkb_rtt_last;       /* Last averaging period */
+
 	char                rkb_name[128];      /* Display name */
 	char                rkb_nodename[128];  /* host:port */
 

=== modified file 'rdkafka_msg.c'
--- rdkafka_msg.c	2013-11-04 16:50:07 +0000
+++ rdkafka_msg.c	2014-01-01 03:04:53 +0000
@@ -64,6 +64,12 @@
 	size_t mlen = sizeof(*rkm);
 	
 	assert(len > 0);
+
+	if (unlikely(len + keylen > rkt->rkt_rk->rk_conf.max_msg_size)) {
+		errno = EMSGSIZE;
+		return -1;
+	}
+
 	if (unlikely(rd_atomic_add(&rkt->rkt_rk->rk_producer.msg_cnt, 1) >
 		     rkt->rkt_rk->rk_conf.queue_buffering_max_msgs)) {
 		rd_atomic_sub(&rkt->rkt_rk->rk_producer.msg_cnt, 1);
@@ -115,7 +121,7 @@
 	int cnt = timedout->rkmq_msg_cnt;
 	
 	/* Assume messages are added in time sequencial order */
-	TAILQ_FOREACH_SAFE(rkm, tmp, &rkmq->rkmq_msgs, rkm_link) {
+	TAILQ_FOREACH_SAFE(rkm, &rkmq->rkmq_msgs, rkm_link, tmp) {
 		if (likely(rkm->rkm_ts_timeout > now))
 			break;
 

=== modified file 'rdkafka_proto.h'
--- rdkafka_proto.h	2013-11-04 16:50:07 +0000
+++ rdkafka_proto.h	2014-01-01 03:04:53 +0000
@@ -81,22 +81,22 @@
 #define RD_KAFKAP_STR_LEN_NULL -1
 /* Returns the actual size of a kafka protocol string representation. */
 #define RD_KAFKAP_STR_SIZE(kstr) (int16_t)(sizeof((kstr)->len) +	\
-					   (ntohs((kstr)->len) ==	\
+					   ((int16_t)ntohs((kstr)->len) == \
 					    RD_KAFKAP_STR_LEN_NULL ?	\
 					    0 : ntohs((kstr)->len)))
 /* Returns the length of the string of a kafka protocol string representation */
 #define RD_KAFKAP_STR_LEN(kstr) (int)((ntohs((kstr)->len) ==		\
 				       RD_KAFKAP_STR_LEN_NULL ?		\
-				       0 : ntohs((kstr)->len)))
+				       0 : (int16_t)ntohs((kstr)->len)))
 
 
 /* Macro suitable for "%.*s" printing. */
 #define RD_KAFKAP_STR_PR(kstr)  \
-	(ntohs((kstr)->len) == RD_KAFKAP_STR_LEN_NULL ?	\
+	((int16_t)ntohs((kstr)->len) == RD_KAFKAP_STR_LEN_NULL ?	\
 	 0 : (int)ntohs((kstr)->len)), (kstr)->str
 
 #define RD_KAFKAP_STR_IS_NULL(kstr) \
-	(ntohs((kstr)->len) == RD_KAFKAP_STR_LEN_NULL)
+	((int16_t)ntohs((kstr)->len) == RD_KAFKAP_STR_LEN_NULL)
 
 static inline int rd_kafkap_str_cmp (const rd_kafkap_str_t *a,
 				     const rd_kafkap_str_t *b) RD_UNUSED;
@@ -137,7 +137,7 @@
 		kstr->len = ntohs(len);
 		memcpy(kstr->str, str, len+1);
 	} else
-		kstr->len = ntohs(RD_KAFKAP_STR_LEN_NULL);
+		kstr->len = (int16_t)ntohs(RD_KAFKAP_STR_LEN_NULL);
 
 	return kstr;
 }
@@ -163,16 +163,17 @@
 #define RD_KAFKAP_BYTES_LEN_NULL -1
 /* Returns the actual size of a kafka protocol bytes representation. */
 #define RD_KAFKAP_BYTES_SIZE(kbytes) (int32_t)(sizeof((kbytes)->len) +	\
-					       (ntohl((kbytes)->len) ==	\
+					       ((int32_t)ntohl((kbytes)->len)==\
 						RD_KAFKAP_BYTES_LEN_NULL ? \
 						0 : ntohl((kbytes)->len)))
 /* Returns the length of the string of a kafka protocol bytes representation */
-#define RD_KAFKAP_BYTES_LEN(kbytes) (int32_t)((ntohl((kbytes)->len) ==	\
+#define RD_KAFKAP_BYTES_LEN(kbytes) (int32_t)(((int32_t)ntohl((kbytes)->len) ==\
 					       RD_KAFKAP_BYTES_LEN_NULL ? \
-					       0 : ntohl((kbytes)->len)))
+					       0 : \
+					       (int32_t)ntohl((kbytes)->len)))
 
 #define RD_KAFKAP_BYTES_IS_NULL(kbytes) \
-	(ntohs((kbytes)->len) == RD_KAFKAP_STR_LEN_NULL)
+	((int32_t)ntohl((kbytes)->len) == RD_KAFKAP_STR_LEN_NULL)
 
 
 static inline int rd_kafkap_bytes_cmp (const rd_kafkap_bytes_t *a,
@@ -207,7 +208,7 @@
 		kbytes->len = ntohl(datalen);
 		memcpy(kbytes->data, data, datalen);
 	} else
-		kbytes->len = ntohl(RD_KAFKAP_BYTES_LEN_NULL);
+		kbytes->len = (int32_t)ntohl(RD_KAFKAP_BYTES_LEN_NULL);
 	
 	return kbytes;
 }

=== modified file 'rdkafka_topic.c'
--- rdkafka_topic.c	2013-11-04 16:50:07 +0000
+++ rdkafka_topic.c	2014-01-01 03:04:53 +0000
@@ -528,7 +528,7 @@
 				   int32_t leader) {
 	rd_kafka_topic_t *rkt;
 	rd_kafka_toppar_t *rktp;
-	rd_kafka_broker_t *rkb;
+	rd_kafka_broker_t *rkb = NULL;
 
 	if (!(rkt = rd_kafka_topic_find(rk, topic))) {
 		rd_kafka_dbg(rk, METADATA, "TOPICUPD",
@@ -537,40 +537,35 @@
 	}
 
 	/* Find broker */
-	rd_kafka_lock(rk);
-	rkb = rd_kafka_broker_find_by_nodeid(rk, leader);
-	rd_kafka_unlock(rk);
-
+	if (leader != -1) {
+		rd_kafka_lock(rk);
+		rkb = rd_kafka_broker_find_by_nodeid(rk, leader);
+		rd_kafka_unlock(rk);
+	}
 
 	rd_kafka_topic_wrlock(rkt);
 
 	rktp = rd_kafka_toppar_get(rkt, partition, 0);
 	assert(rktp);
 
-	if (leader == -1) {
-		/* Topic lost its leader */
-		rd_kafka_toppar_broker_delegate(rktp, NULL);
-		rd_kafka_topic_unlock(rkt);
-
-		/* Query for the topic leader (async) */
-		rd_kafka_topic_leader_query(rk, rkt);
-
-		rd_kafka_toppar_destroy(rktp); /* from get() */
-		rd_kafka_topic_destroy(rkt); /* from find() */
-		return;
-	}
-
-
 	if (!rkb) {
-		rd_kafka_log(rk, LOG_NOTICE, "TOPICBRK",
-			     "Topic %s [%"PRId32"] migrated to unknown "
-			     "broker %"PRId32": requesting metadata update",
-			     topic, partition, leader);
+		int had_leader = rktp->rktp_leader ? 1 : 0;
+
+		if (leader == -1)
+			/* Topic lost its leader */;
+		else
+			rd_kafka_log(rk, LOG_NOTICE, "TOPICBRK",
+				     "Topic %s [%"PRId32"] migrated to unknown "
+				     "broker %"PRId32": "
+				     "requesting metadata update",
+				     topic, partition, leader);
+
 		rd_kafka_toppar_broker_delegate(rktp, NULL);
 		rd_kafka_topic_unlock(rkt);
 
 		/* Query for the topic leader (async) */
-		rd_kafka_topic_leader_query(rk, rkt);
+		if (had_leader)
+			rd_kafka_topic_leader_query(rk, rkt);
 
 		rd_kafka_toppar_destroy(rktp); /* from get() */
 		rd_kafka_topic_destroy(rkt); /* from find() */
@@ -804,7 +799,7 @@
 	cnt = uas.rkmq_msg_cnt;
 	rd_kafka_toppar_unlock(rktp_ua);
 
-	TAILQ_FOREACH_SAFE(rkm, tmp, &uas.rkmq_msgs, rkm_link) {
+	TAILQ_FOREACH_SAFE(rkm, &uas.rkmq_msgs, rkm_link, tmp) {
 		if (unlikely(rd_kafka_msg_partitioner(rkt, NULL, rkm) == -1)) {
 			/* Desired partition not available */
 			rd_kafka_msgq_enq(&failed, rkm);

=== modified file 'rdsysqueue.h'
--- rdsysqueue.h	2013-11-04 16:50:07 +0000
+++ rdsysqueue.h	2014-01-01 03:04:53 +0000
@@ -125,13 +125,6 @@
         (*(((struct headname *)((elm)->field.tqe_prev))->tqh_last))
 #endif
 
-#ifdef TAILQ_FOREACH_SAFE
-#ifdef __APPLE__
-/* Apple's .._SAFE macro has the temporary variable at the end. */
-#undef TAILQ_FOREACH_SAFE
-#endif
-#endif
-
 #ifndef TAILQ_FOREACH_SAFE
 /*
  * TAILQ_FOREACH_SAFE() provides a traversal where the current iterated element
@@ -139,7 +132,7 @@
  * It does not allow freeing or modifying any other element in the list,
  * at least not the next element.
  */
-#define TAILQ_FOREACH_SAFE(elm,tmpelm,head,field)			\
+#define TAILQ_FOREACH_SAFE(elm,head,field,tmpelm)			\
 	for ((elm) = TAILQ_FIRST(head) ;				\
 	     (elm) && ((tmpelm) = TAILQ_NEXT((elm), field), 1) ;	\
 	     (elm) = (tmpelm))

=== added file 'tests/0003-msgmaxsize.c'
--- tests/0003-msgmaxsize.c	1970-01-01 00:00:00 +0000
+++ tests/0003-msgmaxsize.c	2014-01-01 03:04:53 +0000
@@ -0,0 +1,169 @@
+/*
+ * librdkafka - Apache Kafka C library
+ *
+ * Copyright (c) 2012-2013, Magnus Edenhill
+ * All rights reserved.
+ * 
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met: 
+ * 
+ * 1. Redistributions of source code must retain the above copyright notice,
+ *    this list of conditions and the following disclaimer. 
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ *    this list of conditions and the following disclaimer in the documentation
+ *    and/or other materials provided with the distribution. 
+ * 
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+/**
+ * Tests "message.bytes.max"
+ * Issue #24
+ */
+
+#define _GNU_SOURCE
+#include <signal.h>
+#include <sys/time.h>
+#include <time.h>
+
+#include "test.h"
+
+/* Typical include path would be <librdkafka/rdkafka.h>, but this program
+ * is built from within the librdkafka source tree and thus differs. */
+#include "rdkafka.h"  /* for Kafka driver */
+
+
+static int msgs_wait = 0; /* bitmask */
+
+/**
+ * Delivery report callback.
+ * Called for each message once to signal its delivery status.
+ */
+static void dr_cb (rd_kafka_t *rk, void *payload, size_t len,
+		   rd_kafka_resp_err_t err, void *opaque, void *msg_opaque) {
+	int msgid = *(int *)msg_opaque;
+
+	free(msg_opaque);
+
+	if (err)
+		TEST_FAIL("Unexpected delivery error for message #%i: %s\n",
+			  msgid, rd_kafka_err2str(err));
+
+	if (!(msgs_wait & (1 << msgid)))
+		TEST_FAIL("Unwanted delivery report for message #%i "
+			  "(waiting for 0x%x)\n", msgid, msgs_wait);
+
+	TEST_SAY("Delivery report for message #%i: %s\n",
+		 msgid, rd_kafka_err2str(err));
+
+	msgs_wait &= ~(1 << msgid);
+}
+
+
+int main (int argc, char **argv) {
+	char *topic = "rdkafkatest1";
+	int partition = 0;
+	int r;
+	rd_kafka_t *rk;
+	rd_kafka_topic_t *rkt;
+	rd_kafka_conf_t *conf;
+	rd_kafka_topic_conf_t *topic_conf;
+	char errstr[512];
+	char msg[100000];
+	int msgcnt = 10;
+	int i;
+
+	/* Socket hangups are gracefully handled in librdkafka on socket error
+	 * without the use of signals, so SIGPIPE should be ignored by the
+	 * calling program. */
+	signal(SIGPIPE, SIG_IGN);
+
+
+	test_conf_init(&conf, &topic_conf, 10);
+
+	/* Set a small maximum message size. */
+	if (rd_kafka_conf_set(conf, "message.max.bytes", "100000",
+			      errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
+		TEST_FAIL("%s\n", errstr);
+
+	/* Set delivery report callback */
+	rd_kafka_conf_set_dr_cb(conf, dr_cb);
+
+	/* Create kafka instance */
+	rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf,
+			  errstr, sizeof(errstr));
+	if (!rk)
+		TEST_FAIL("Failed to create rdkafka instance: %s\n", errstr);
+
+	TEST_SAY("Created    kafka instance %s\n", rd_kafka_name(rk));
+
+	rkt = rd_kafka_topic_new(rk, topic, topic_conf);
+	if (!rkt)
+		TEST_FAIL("Failed to create topic: %s\n",
+			  strerror(errno));
+
+	memset(msg, 0, sizeof(msg));
+
+	/* Produce 'msgcnt' messages, size odd ones larger than max.bytes,
+	 * and even ones smaller than max.bytes. */
+	for (i = 0 ; i < msgcnt ; i++) {
+		int *msgidp = malloc(sizeof(*msgidp));
+		size_t len;
+		int toobig = i & 1;
+
+		*msgidp = i;
+		if (toobig) {
+			/* Too big */
+			len = 200000;
+		} else {
+			/* Good size */
+			len = 5000;
+			msgs_wait |= (1 << i);
+		}
+
+		snprintf(msg, sizeof(msg), "%s test message #%i", argv[0], i);
+		r = rd_kafka_produce(rkt, partition, RD_KAFKA_MSG_F_COPY,
+				     msg, len, NULL, 0, msgidp);
+
+		if (toobig) {
+			if (r != -1)
+				TEST_FAIL("Succeeded to produce too "
+					  "large message #%i\n", i);
+			free(msgidp);
+		} else if (r == -1)
+			TEST_FAIL("Failed to produce message #%i: %s\n",
+				  i, strerror(errno));
+	}
+
+	/* Wait for messages to be delivered. */
+	while (rd_kafka_outq_len(rk) > 0)
+		rd_kafka_poll(rk, 50);
+
+	if (msgs_wait != 0)
+		TEST_FAIL("Still waiting for messages: 0x%x\n", msgs_wait);
+
+	/* Destroy topic */
+	rd_kafka_topic_destroy(rkt);
+		
+	/* Destroy rdkafka instance */
+	TEST_SAY("Destroying kafka instance %s\n", rd_kafka_name(rk));
+	rd_kafka_destroy(rk);
+
+	/* Wait for everything to be cleaned up since broker destroys are
+	 * handled in its own thread. */
+	test_wait_exit(10);
+
+	/* If we havent failed at this point then
+	 * there were no threads leaked */
+	return 0;
+}

=== modified file 'tests/Makefile'
--- tests/Makefile	2013-11-04 16:50:07 +0000
+++ tests/Makefile	2014-01-01 03:04:53 +0000
@@ -1,4 +1,4 @@
-TESTS ?= 0001-multiobj.test 0002-msgtimeout.test
+TESTS ?= 0001-multiobj.test 0002-msgtimeout.test 0003-msgmaxsize.test
 CC ?= cc
 CXX ?= g++
 CFLAGS += -g -Wall -Werror -Wfloat-equal -Wpointer-arith -O2 -I../