avro: out_kafka: fix a bug in avro schema_id

Signed-off-by: jh.park <jh.park@bucketplace.net>

Author: siji-on

include/fluent-bit/flb_avro.h

@@ -31,7 +31,7 @@
 #define MEMORY_POOL_MINIMUM_SIZE sizeof(void *)
 
 struct flb_avro_fields {
-    flb_sds_t schema_id;
+    int schema_id;
     flb_sds_t schema_str;
 };
 

plugins/out_kafka/kafka.c

@@ -312,29 +312,29 @@ int produce_message(struct flb_time *tm, msgpack_object *map,
 #ifdef FLB_HAVE_AVRO_ENCODER
     else if (ctx->format == FLB_KAFKA_FMT_AVRO) {
 
-        flb_plg_debug(ctx->ins, "avro schema ID:%s:\n", ctx->avro_fields.schema_id);
+        flb_plg_debug(ctx->ins, "avro schema ID:%d:\n", ctx->avro_fields.schema_id);
         flb_plg_debug(ctx->ins, "avro schema string:%s:\n", ctx->avro_fields.schema_str);
 
 	// if there's no data then log it and return
         if (mp_sbuf.size == 0) {
-            flb_plg_error(ctx->ins, "got zero bytes decoding to avro AVRO:schemaID:%s:\n", ctx->avro_fields.schema_id);
+            flb_plg_error(ctx->ins, "got zero bytes decoding to avro AVRO:schemaID:%d:\n", ctx->avro_fields.schema_id);
             msgpack_sbuffer_destroy(&mp_sbuf);
             return FLB_OK;
         }
 
 	// is the line is too long log it and return
         if (mp_sbuf.size > AVRO_LINE_MAX_LEN) {
-            flb_plg_warn(ctx->ins, "skipping long line AVRO:len:%zu:limit:%zu:schemaID:%s:\n", (size_t)mp_sbuf.size, (size_t)AVRO_LINE_MAX_LEN, ctx->avro_fields.schema_id);
+            flb_plg_warn(ctx->ins, "skipping long line AVRO:len:%zu:limit:%zu:schemaID:%d:\n", (size_t)mp_sbuf.size, (size_t)AVRO_LINE_MAX_LEN, ctx->avro_fields.schema_id);
             msgpack_sbuffer_destroy(&mp_sbuf);
             return FLB_OK;
         }
 
-        flb_plg_debug(ctx->ins, "using default buffer AVRO:len:%zu:limit:%zu:schemaID:%s:\n", (size_t)mp_sbuf.size, (size_t)AVRO_DEFAULT_BUFFER_SIZE, ctx->avro_fields.schema_id);
+        flb_plg_debug(ctx->ins, "using default buffer AVRO:len:%zu:limit:%zu:schemaID:%d:\n", (size_t)mp_sbuf.size, (size_t)AVRO_DEFAULT_BUFFER_SIZE, ctx->avro_fields.schema_id);
         out_buf = avro_buff;
         out_size = AVRO_DEFAULT_BUFFER_SIZE;
 
 	if (mp_sbuf.size + AVRO_SCHEMA_OVERHEAD >= AVRO_DEFAULT_BUFFER_SIZE) {
-            flb_plg_info(ctx->ins, "upsizing to dynamic buffer AVRO:len:%zu:schemaID:%s:\n", (size_t)mp_sbuf.size, ctx->avro_fields.schema_id);
+            flb_plg_info(ctx->ins, "upsizing to dynamic buffer AVRO:len:%zu:schemaID:%d:\n", (size_t)mp_sbuf.size, ctx->avro_fields.schema_id);
             avro_fast_buffer = false;
             // avro will always be  smaller than msgpack
             // it contains no meta-info aside from the schemaid
@@ -344,14 +344,14 @@ int produce_message(struct flb_time *tm, msgpack_object *map,
             out_size = mp_sbuf.size + AVRO_SCHEMA_OVERHEAD;
             out_buf = flb_malloc(out_size);
             if (!out_buf) {
-                flb_plg_error(ctx->ins, "error allocating memory for decoding to AVRO:schema:%s:schemaID:%s:\n", ctx->avro_fields.schema_str, ctx->avro_fields.schema_id);
+                flb_plg_error(ctx->ins, "error allocating memory for decoding to AVRO:schema:%s:schemaID:%d:\n", ctx->avro_fields.schema_str, ctx->avro_fields.schema_id);
                 msgpack_sbuffer_destroy(&mp_sbuf);
                 return FLB_ERROR;
             }
 	}
 
         if(!flb_msgpack_raw_to_avro_sds(mp_sbuf.data, mp_sbuf.size, &ctx->avro_fields, out_buf, &out_size)) {
-            flb_plg_error(ctx->ins, "error encoding to AVRO:schema:%s:schemaID:%s:\n", ctx->avro_fields.schema_str, ctx->avro_fields.schema_id);
+            flb_plg_error(ctx->ins, "error encoding to AVRO:schema:%s:schemaID:%d:\n", ctx->avro_fields.schema_str, ctx->avro_fields.schema_id);
             msgpack_sbuffer_destroy(&mp_sbuf);
             if (!avro_fast_buffer) {
                 flb_free(out_buf);
@@ -639,8 +639,8 @@ static struct flb_config_map config_map[] = {
     "Set AVRO schema."
    },
    {
-    FLB_CONFIG_MAP_STR, "schema_id", (char *)NULL,
-    0, FLB_FALSE, 0,
+    FLB_CONFIG_MAP_INT, "schema_id", (char *)NULL,
+    0, FLB_TRUE, offsetof(struct flb_out_kafka, avro_fields) + offsetof(struct flb_avro_fields, schema_id),
     "Set AVRO schema ID."
    },
 #endif

plugins/out_kafka/kafka_config.c

@@ -249,10 +249,6 @@ struct flb_out_kafka *flb_out_kafka_create(struct flb_output_instance *ins,
     if (tmp) {
         ctx->avro_fields.schema_str = flb_sds_create(tmp);
     }
-    tmp = flb_output_get_property("schema_id", ins);
-    if (tmp) {
-        ctx->avro_fields.schema_id = flb_sds_create(tmp);
-    }
 #endif
 
     /* Config: Topic */
@@ -282,7 +278,7 @@ struct flb_out_kafka *flb_out_kafka_create(struct flb_output_instance *ins,
 
     flb_plg_info(ctx->ins, "brokers='%s' topics='%s'", ctx->kafka.brokers, tmp);
 #ifdef FLB_HAVE_AVRO_ENCODER
-    flb_plg_info(ctx->ins, "schemaID='%s' schema='%s'", ctx->avro_fields.schema_id, ctx->avro_fields.schema_str);
+    flb_plg_info(ctx->ins, "schemaID='%d' schema='%s'", ctx->avro_fields.schema_id, ctx->avro_fields.schema_str);
 #endif
 
     return ctx;
@@ -324,7 +320,6 @@ int flb_out_kafka_destroy(struct flb_out_kafka *ctx)
 
 #ifdef FLB_HAVE_AVRO_ENCODER
     // avro
-    flb_sds_destroy(ctx->avro_fields.schema_id);
     flb_sds_destroy(ctx->avro_fields.schema_str);
 #endif
 

src/flb_avro.c

@@ -268,7 +268,7 @@ bool flb_msgpack_raw_to_avro_sds(const void *in_buf, size_t in_size, struct flb_
 
     avro_writer_t awriter;
     flb_debug("in flb_msgpack_raw_to_avro_sds\n");
-    flb_debug("schemaID:%s:\n", ctx->schema_id);
+    flb_debug("schemaID:%d:\n", ctx->schema_id);
     flb_debug("schema string:%s:\n", ctx->schema_str);
 
     size_t schema_json_len = flb_sds_len(ctx->schema_str);
@@ -341,18 +341,15 @@ bool flb_msgpack_raw_to_avro_sds(const void *in_buf, size_t in_size, struct flb_
     }
 
     // write the schemaid
-    // its md5hash of the avro schema
-    // it looks like this c4b52aaf22429c7f9eb8c30270bc1795
-    const char *pos = ctx->schema_id;
-    unsigned char val[16];
-    size_t count;
-    for (count = 0; count < sizeof val/sizeof *val; count++) {
-            sscanf(pos, "%2hhx", &val[count]);
-            pos += 2;
-    }
+    unsigned int id = ctx->schema_id;
+    unsigned char val[4];
+    val[0] = (id >> 24) & 0xFF;
+    val[1] = (id >> 16) & 0xFF;
+    val[2] = (id >> 8) & 0xFF;
+    val[3] = id & 0xFF;
 
     // write it into a buffer which can be passed to librdkafka
-    rval = avro_write(awriter, val, 16);
+    rval = avro_write(awriter, val, 4);
     if (rval != 0) {
         flb_error("Unable to write schemaid\n");
         avro_writer_free(awriter);