retry when metadata empty

@@ -459,7 +459,8 @@ int SrsKafkaProducer::request_metadata()
                   senabled.c_str(), sbrokers.c_str(), lb->current(), server.c_str(), port, topic.c_str());
     }
     
-    // connect to kafka server.
+    // reconnect to kafka server.
+    transport->close();
     if ((ret = transport->connect(server, port, SRS_CONSTS_KAFKA_TIMEOUT_US)) != ERROR_SUCCESS) {
         srs_error("kafka connect %s:%d failed. ret=%d", server.c_str(), port, ret);
         return ret;
@@ -473,6 +474,16 @@ int SrsKafkaProducer::request_metadata()
     }
     SrsAutoFree(SrsKafkaTopicMetadataResponse, metadata);
     
+    // we may need to request multiple times.
+    // for example, the first time to create a none-exists topic, then query metadata.
+    if (!metadata->metadatas.empty()) {
+        SrsKafkaTopicMetadata* topic = metadata->metadatas.at(0);
+        if (topic->metadatas.empty()) {
+            srs_warn("topic %s metadata empty, retry.", topic->name.to_str().c_str());
+            return ret;
+        }
+    }
+    
     // show kafka metadata.
     string summary = srs_kafka_metadata_summary(metadata);
     srs_trace("kafka metadata: %s", summary.c_str());

@@ -152,6 +152,10 @@ public:
     {
         return length;
     }
+    virtual bool empty()
+    {
+        return elems.empty();
+    }
     virtual T* at(int index)
     {
         return elems.at(index);
@@ -241,6 +245,10 @@ public:
     {
         return length;
     }
+    virtual bool empty()
+    {
+        return elems.empty();
+    }
     virtual int32_t at(int index)
     {
         return elems.at(index);