Merge pull request #168 from farsightsec/next

Release 1.3.1 to depend on librdkafka error handling.

Author: alesage

ChangeLog

@@ -1,3 +1,7 @@
+nmsg (1.3.1-1)
+
+  * Rely on built-in librdkafka error handling.
+
 nmsg (1.2.0)
 
   * Add support for Kafka i/o in JSON and binary forms.

configure.ac

@@ -2,7 +2,7 @@ AC_PREREQ(2.64)
 
 m4_define(nmsg_major_version, 1)
 m4_define(nmsg_minor_version, 3)
-m4_define(nmsg_patchlevel_version, 0)
+m4_define(nmsg_patchlevel_version, 1)
 m4_define(nmsg_version,
 	  nmsg_major_version.nmsg_minor_version.nmsg_patchlevel_version)
 m4_define(nmsg_version_number,

debian/changelog

@@ -1,3 +1,9 @@
+nmsg (1.3.1-1) debian-fsi; urgency=medium
+
+  * Rely on built-in librdkafka error handling.
+
+ -- Farsight Security, Inc. <software@farsightsecurity.com>  Thu, 31 Oct 2024 18:19:04 +0000
+
 nmsg (1.3.0-1) debian-fsi; urgency=medium
 
   * Added external configuration for Kafka IO.

nmsg.spec

@@ -1,5 +1,5 @@
 Name:           nmsg
-Version:        1.3.0
+Version:        1.3.1
 Release:        1%{?dist}
 Summary:	network message encapsulation library
 

nmsg/kafkaio.c

@@ -210,7 +210,7 @@ _kafka_state_to_str(kafka_state state)
 static void
 _kafka_set_state(kafka_ctx_t ctx, const char *func, kafka_state state)
 {
-	_nmsg_dprintf(3, "%s changing state from %s to %s\n", func,
+	_nmsg_dprintf(3, "%s: changing state from %s to %s\n", func,
 		_kafka_state_to_str(ctx->state), _kafka_state_to_str(state));
 	ctx->state = state;
 }
@@ -597,19 +597,15 @@ _kafka_error_cb(rd_kafka_t *rk, int err, const char *reason, void *opaque)
 		return;
 	}
 	switch (err_kafka) {
-		/* Keep retrying on socket disconnect, brokers down and message timeout */
 		case RD_KAFKA_RESP_ERR__TRANSPORT:
 		case RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN:
 		case RD_KAFKA_RESP_ERR__MSG_TIMED_OUT:
-			_nmsg_dprintf(2, "%s: got Kafka error %d: %s\n", __func__, err, reason);
-			break;
 		case RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION:
 		case RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART:
 		case RD_KAFKA_RESP_ERR_OFFSET_OUT_OF_RANGE:
-		/* At the moment treat any broker's error as fatal */
 		default:
+			/* Just log, let librdkafka handle all errors */
 			_nmsg_dprintf(2, "%s: got Kafka error %d: %s\n", __func__, err, reason);
-			_kafka_set_state(ctx, __func__, kafka_state_break);
 			break;
 	}
 }
@@ -630,10 +626,7 @@ _kafka_delivery_cb(rd_kafka_t *rk, const rd_kafka_message_t *rkmessage, void *op
 	}
 	if (rkmessage->err != RD_KAFKA_RESP_ERR_NO_ERROR) {
 		int level = 2;
-		if (rkmessage->err != RD_KAFKA_RESP_ERR__MSG_TIMED_OUT) {
-			_kafka_set_state(ctx, __func__, kafka_state_break);
-			rd_kafka_yield(rk);
-		} else {
+		if (rkmessage->err == RD_KAFKA_RESP_ERR__MSG_TIMED_OUT) {
 			ctx->dropped++;
 			level = 4;
 		}