gfwleak/tango-tfe
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

TSG-7471 Proxy连接kafka时增加认证信息

Browse Source
This commit is contained in:
fengweihao
2021-08-19 16:24:19 +08:00
parent 2a729a48b3
commit c41a67ca2b
5 changed files with 46 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
common/src/tfe_kafka_logger.cpp
View File

@@ -34,7 +34,7 @@ error:
return INADDR_NONE;
}
static rd_kafka_t *create_kafka_handle(const char *brokerlist, void *local_logger)
static rd_kafka_t *create_kafka_handle(const char *brokerlist, const char *sasl_username, const char *sasl_passwd, void *local_logger)
{
int ret;
char kafka_errstr[1024] = {0};
@@ -65,6 +65,26 @@ static rd_kafka_t *create_kafka_handle(const char *brokerlist, void *local_logge
return NULL;
}
if (strlen(sasl_username) > 0 && strlen(sasl_passwd) > 0)
{
rd_kafka_conf_set(rconf, "security.protocol", "sasl_plaintext", kafka_errstr, sizeof(kafka_errstr));
rd_kafka_conf_set(rconf, "sasl.mechanisms", "PLAIN", kafka_errstr, sizeof(kafka_errstr));
ret = rd_kafka_conf_set(rconf, "sasl.username", sasl_username, kafka_errstr, sizeof(kafka_errstr));
if (ret != RD_KAFKA_CONF_OK)
{
TFE_LOG_ERROR(local_logger, "Error to set kafka \"sasl.username\", %s.", kafka_errstr);
rd_kafka_conf_destroy(rconf);
return NULL;
}
ret = rd_kafka_conf_set(rconf, "sasl.password", sasl_passwd, kafka_errstr, sizeof(kafka_errstr));
if (ret != RD_KAFKA_CONF_OK)
{
TFE_LOG_ERROR(local_logger, "Error to set kafka \"sasl.password\", %s.", kafka_errstr);
rd_kafka_conf_destroy(rconf);
return NULL;
}
}
//The conf object is freed by this function and must not be used or destroyed by the application sub-sequently.
handle = rd_kafka_new(RD_KAFKA_PRODUCER, rconf, kafka_errstr, sizeof(kafka_errstr));
rconf = NULL;
@@ -84,7 +104,7 @@ static rd_kafka_t *create_kafka_handle(const char *brokerlist, void *local_logge
return handle;
}
tfe_kafka_logger_t *tfe_kafka_logger_create(int enable, const char *nic_name, const char *brokerlist, const char *topic_name, void *local_logger)
tfe_kafka_logger_t *tfe_kafka_logger_create(int enable, const char *nic_name, const char *brokerlist, const char *topic_name, const char *sasl_username, const char *sasl_passwd, void *local_logger)
{
tfe_kafka_logger_t *logger = (tfe_kafka_logger_t *)calloc(1, sizeof(tfe_kafka_logger_t));
if (!logger)
@@ -104,7 +124,7 @@ tfe_kafka_logger_t *tfe_kafka_logger_create(int enable, const char *nic_name, co
inet_ntop(AF_INET, &(logger->local_ip_num), logger->local_ip_str, sizeof(logger->local_ip_str));
strncpy(logger->broker_list, brokerlist, strlen(brokerlist));
logger->kafka_handle = create_kafka_handle(logger->broker_list, local_logger);
logger->kafka_handle = create_kafka_handle(logger->broker_list, sasl_username, sasl_passwd, local_logger);
if (logger->kafka_handle == NULL)
{
TFE_LOG_ERROR(local_logger, "Error to creat kafka handler with brokerlist: %s.", logger->broker_list);