Merge pull request ceph#51864 from yuvalif/wip-yuval-fix-61540

rgw/kafka: make sure that destroy is called after connection is removed

Author: yuvalif

src/rgw/rgw_kafka.cc

@@ -100,10 +100,12 @@ struct connection_t {
     // fire all remaining callbacks (if not fired by rd_kafka_flush)
     std::for_each(callbacks.begin(), callbacks.end(), [this](auto& cb_tag) {
         cb_tag.cb(status);
-        ldout(cct, 20) << "Kafka destroy: invoking callback with tag=" << cb_tag.tag << dendl;
+        ldout(cct, 20) << "Kafka destroy: invoking callback with tag=" << cb_tag.tag << 
+          " for: " << broker << dendl;
       });
     callbacks.clear();
     delivery_tag = 1;
+    ldout(cct, 20) << "Kafka destroy: complete for: " << broker << dendl;
   }
 
   bool is_ok() const {
@@ -459,7 +461,6 @@ class Manager {
         // Checking the connection idlesness
         if(conn->timestamp.sec() + max_idle_time < ceph_clock_now()) {
           ldout(conn->cct, 20) << "kafka run: deleting a connection due to idle behaviour: " << ceph_clock_now() << dendl;
-          conn->destroy(STATUS_CONNECTION_IDLE);
           std::lock_guard lock(connections_lock);
           conn_it = connections.erase(conn_it);
           --connection_count; \