Why Disconnected: veat security.protocol is correctly configured, broker might require SASL authentication?

[suitgo]

I am testing my log for produce information send to kafka brokers, the producer use the librdkafka.so for produce, when the producer send message to broker, I got the following errors. The broker security protocol was set plaintext, and I have checked that the producer also set as plaintext. I can not find out which conf was set as wrong way, please help me. Thanks.
The error log:
%3|1636424579.709|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: hdp03.pcl-test.com:6667/1002: Disconnected: veat security.protocol is correctly configured, broker might require SASL authentication (after 55ms in state UP, 4 identica(s) suppressed)
%6|1636424616.991|FAIL|rdkafka#producer-1| [thrd:hdp03.pcl-test.com:6667/1002]: hdp03.pcl-test.com:6667/1002: Disconnectedy that security.protocol is correctly configured, broker might require SASL authentication (after 50ms in state UP, 4 idenrror(s) suppressed)

Info:

1. Linux version: Linux version 3.10.0-862.el7.x86_64 (builder@kbuilder.dev.centos.org) (gcc version 4.8.5 20150623 (Red Hat 4.8.5-28) (GCC) ) Bugfix in example code #1 SMP Fri Apr 20 16:44:24 UTC 2018

2. GCC version: 4.8.5 20150623

3. Producer code:
   ...// create produer
   {
   rd_kafka_conf_t *conf;
   char errstr[512];

   conf = rd_kafka_conf_new();
   if (rd_kafka_conf_set(conf, "bootstrap.servers", brokers.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
   printf("Producer: bootstrap.servers set fial for:%s\n", errstr);
   return;
   }

   rd_kafka_conf_set_dr_msg_cb(conf, PCL_KafkaDrmsgCb);

   g_kafkaProducer = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
   if (g_kafkaProducer == NULL) {
   printf("rd_kafka_new fail for:%s\n", errstr);
   return;
   }
   g_kafkaProducerTopic = rd_kafka_topic_new(g_kafkaProducer, topic.c_str(), NULL);
   }

...// send queue
{
UINT32 count = 0;
UINT32 alllength = 0;
UINT32 loglen = 0;
while (true) {
if (queue.empty()) {
break;
}

	string &loginfo = queue.front();
    loglen = loginfo.length();
    alllength = loglen + 1;
    KafkaProducerSendMsg((UINT8*)loginfo.c_str(), alllength);
    queue.pop();

    count++;
    if (count >= 100) {
        break;
    }
}
if (count == 0) {
    usleep(10000);
}
usleep(10);

}
VOID KafkaProducerSendMsg(UINT8 *buf, UINT32 length)
{
rd_kafka_resp_err_t err;

while (true) {
    err = rd_kafka_producev(
            g_kafkaProducer,
            RD_KAFKA_V_RKT(g_kafkaProducerTopic),
            RD_KAFKA_V_PARTITION(RD_KAFKA_PARTITION_UA),
            RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY),
            RD_KAFKA_V_VALUE(buf, length),
            RD_KAFKA_V_OPAQUE(NULL),
            RD_KAFKA_V_END);

    if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
        printf("Failed to produce reason:%s\n", rd_kafka_err2str(err));
        if (err == RD_KAFKA_RESP_ERR__QUEUE_FULL) {
            rd_kafka_poll(g_kafkaProducer, 1000);
            continue;
        }
    }
    rd_kafka_poll(g_kafkaProducer, -1);
    break;
}

}

Thanks a lot.

[suitgo]

Add informaiton the librdkafka version: 1.8.2(release version)