Prevent broker delegation from getting stuck

Author: kphelps

src/rdkafka_broker.c

@@ -3327,6 +3327,7 @@ rd_kafka_broker_op_serve(rd_kafka_broker_t *rkb, rd_kafka_op_t *rko) {
                 rkb->rkb_toppar_cnt++;
                 rd_kafka_broker_unlock(rkb);
                 rktp->rktp_broker = rkb;
+                rktp->rktp_broker_id = rkb->rkb_nodeid;
                 rd_assert(!rktp->rktp_msgq_wakeup_q);
                 rktp->rktp_msgq_wakeup_q = rd_kafka_q_keep(rkb->rkb_ops);
                 rd_kafka_broker_keep(rkb);
@@ -3430,6 +3431,7 @@ rd_kafka_broker_op_serve(rd_kafka_broker_t *rkb, rd_kafka_op_t *rko) {
                         rktp->rktp_msgq_wakeup_q = NULL;
                 }
                 rktp->rktp_broker = NULL;
+                rktp->rktp_broker_id = -1;
 
                 rd_assert(rktp->rktp_flags & RD_KAFKA_TOPPAR_F_ON_RKB);
                 rktp->rktp_flags &= ~RD_KAFKA_TOPPAR_F_ON_RKB;

src/rdkafka_topic.c

@@ -605,17 +605,14 @@ int rd_kafka_toppar_broker_update(rd_kafka_toppar_t *rktp,
                                   int32_t broker_id,
                                   rd_kafka_broker_t *rkb,
                                   const char *reason) {
-
-        rktp->rktp_broker_id = broker_id;
-
         if (!rkb) {
                 int had_broker = rktp->rktp_broker ? 1 : 0;
                 rd_kafka_toppar_broker_delegate(rktp, NULL);
                 return had_broker ? -1 : 0;
         }
 
         if (rktp->rktp_broker) {
-                if (rktp->rktp_broker == rkb) {
+                if (rktp->rktp_broker == rkb && !rktp->rktp_next_broker) {
                         /* No change in broker */
                         return 0;
                 }