1 files changed, 47 insertions, 22 deletions

diff --git a/common/src/tfe_kafka_logger.cpp b/common/src/tfe_kafka_logger.cpp
index 26cfea6..28f4ce8 100644
--- a/common/src/tfe_kafka_logger.cpp
+++ b/common/src/tfe_kafka_logger.cpp
@@ -34,7 +34,7 @@ error:
     return INADDR_NONE;
 }
 
-static rd_kafka_t *create_kafka_handle(const char *brokerlist, const char *sasl_username, const char *sasl_passwd, void *local_logger)
+static rd_kafka_t *create_kafka_handle(const char *brokerlist, const char *sasl_username, const char *sasl_passwd, const char *topic_name, void *local_logger)
 {
     int ret;
     char kafka_errstr[1024] = {0};
@@ -64,6 +64,13 @@ static rd_kafka_t *create_kafka_handle(const char *brokerlist, const char *sasl_
         rd_kafka_conf_destroy(rconf);
         return NULL;
     }
+	ret = rd_kafka_conf_set(rconf, "client.id", topic_name, kafka_errstr, sizeof(kafka_errstr));
+    if (ret != RD_KAFKA_CONF_OK)
+    {
+        TFE_LOG_ERROR(local_logger, "Error to set kafka \"client.id\", %s.", kafka_errstr);
+        rd_kafka_conf_destroy(rconf);
+        return NULL;
+    }
 
 	if (strlen(sasl_username) > 0 && strlen(sasl_passwd) > 0)
 	{
@@ -109,11 +116,11 @@ int tfe_kafka_logger_topic_new(tfe_kafka_logger_t *logger, const char *topic_nam
     if(logger && logger->enable)
     {
         strncpy(logger->topic_name[topic_id], topic_name, sizeof(logger->topic_name[topic_id])-1);
-        logger->kafka_topic[topic_id] = rd_kafka_topic_new(logger->kafka_handle, topic_name, NULL);
+        logger->kafka_topic[topic_id] = rd_kafka_topic_new(logger->kafka_handle[topic_id], topic_name, NULL);
         if (logger->kafka_topic[topic_id] == NULL)
         {
             TFE_LOG_ERROR(local_logger, "Error to creat kafka topic: %s.", topic_name);
-            rd_kafka_destroy(logger->kafka_handle);
+            rd_kafka_destroy(logger->kafka_handle[topic_id]);
             free(logger);
             return -1;
         }
@@ -121,22 +128,27 @@ int tfe_kafka_logger_topic_new(tfe_kafka_logger_t *logger, const char *topic_nam
     return 0;
 }
 
-tfe_kafka_logger_t *tfe_kafka_logger_create(int enable, const char *nic_name, const char *brokerlist, const char *sasl_username, const char *sasl_passwd, void *local_logger)
+tfe_kafka_logger_t *tfe_kafka_logger_create(int enable, const char *nic_name, const char *brokerlist, void *local_logger)
 {
 	char *override_sled_ip=NULL;
+
     tfe_kafka_logger_t *logger = (tfe_kafka_logger_t *)calloc(1, sizeof(tfe_kafka_logger_t));
     if (!logger)
-        return NULL;
+    {
+		return NULL;
+	}
 
     logger->enable = enable;
     if (!logger->enable)
-        return logger;
+    {
+		return logger;
+	}
 
     override_sled_ip = getenv("OVERRIDE_SLED_IP");
 	if(override_sled_ip != NULL)
 	{
 		strncpy(logger->local_ip_str, override_sled_ip, sizeof(logger->local_ip_str)-1);
-		goto create_kafka;
+		goto finish;
 	}
 
 	logger->local_ip_num = get_ip_by_eth_name(nic_name);
@@ -147,32 +159,45 @@ tfe_kafka_logger_t *tfe_kafka_logger_create(int enable, const char *nic_name, co
 	    return NULL;
 	}
 	inet_ntop(AF_INET, &(logger->local_ip_num), logger->local_ip_str, sizeof(logger->local_ip_str));
+finish:
+	strncpy(logger->broker_list, brokerlist, sizeof(logger->broker_list)-1);
+	return logger;
+}
 
-create_kafka:
-    strncpy(logger->broker_list, brokerlist, sizeof(logger->broker_list)-1);
-    logger->kafka_handle = create_kafka_handle(logger->broker_list, sasl_username, sasl_passwd, local_logger);
-    if (logger->kafka_handle == NULL)
+int tfe_logger_create_kafka_topic(tfe_kafka_logger_t *logger, const char *sasl_username, const char *sasl_passwd, const char *topic_name, int topic_id, void *local_logger)
+{
+    if(!logger->enable)
+    {
+        return 0;
+    }
+
+	logger->kafka_handle[topic_id] = create_kafka_handle(logger->broker_list, sasl_username, sasl_passwd, topic_name, local_logger);
+    if (logger->kafka_handle[topic_id] == NULL)
     {
         TFE_LOG_ERROR(local_logger, "Error to creat kafka handler with brokerlist: %s.", logger->broker_list);
         free(logger);
-        return NULL;
+        return -1;
     }
-    return logger;
+	tfe_kafka_logger_topic_new(logger, topic_name, topic_id, logger);
+	return 0;
 }
 
 void tfe_kafka_logger_destroy(tfe_kafka_logger_t *logger)
 {
     if (logger)
     {
-        if (logger->kafka_handle)
-            rd_kafka_destroy(logger->kafka_handle);
-
-        if (logger->kafka_topic[TOPIC_LOGGER])
-            rd_kafka_topic_destroy(logger->kafka_topic[TOPIC_LOGGER]);
-
-        if (logger->kafka_topic[TOPIC_BUCKET])
-            rd_kafka_topic_destroy(logger->kafka_topic[TOPIC_BUCKET]);
-
+		for(int i=0; i<TOPIC_MAX; i++)
+		{
+			if(logger->kafka_topic[i])
+			{
+				rd_kafka_topic_destroy(logger->kafka_topic[i]);
+			}
+
+			if(logger->kafka_handle[i])
+			{
+				rd_kafka_destroy(logger->kafka_handle[i]);
+			}
+		}
         free(logger);
         logger = NULL;
     }