[KIP-848] Use metadata cache by topic id, fixes and improvements (#4661)

- rename 'generic' protocol to 'classic'
- consumer group serve timer to awake the loop earlier
- compare and find into topic partition list by topic id only
- fix memory leak when instance creation fails and app_conf is provided
- fix cases where HB response is received after unsubscription
- use topic name from current assignment if it's missing from metadata
- expedite heartbeat simplification and
    fixes to next interval check and to the place where
    number of retries is increased
- expedite HB after changing state back to INIT
- use the CONSUMER_F_WAIT_REJOIN to trigger
  the rejoin instead of calling it from max poll interval timer
- schedule timer for next execution
  reschedule it if expected earlier
  expedite through scheduling cgrp serve timer
- treat unsupported feature error
   as fatal
- avoid removing partitions not matched by a new subscription
   immediately, for a possible desynchronization with the coordinator, 
   with said partitions not being consumed anymore

Author: emasab

CONFIGURATION.md

@@ -109,7 +109,9 @@ group.instance.id                        |  C  |                 |
 partition.assignment.strategy            |  C  |                 | range,roundrobin | medium     | The name of one or more partition assignment strategies. The elected group leader will use a strategy supported by all members of the group to assign partitions to group members. If there is more than one eligible strategy, preference is determined by the order of this list (strategies earlier in the list have higher priority). Cooperative and non-cooperative (eager) strategies must not be mixed. Available strategies: range, roundrobin, cooperative-sticky. <br>*Type: string*
 session.timeout.ms                       |  C  | 1 .. 3600000    |         45000 | high       | Client group session and failure detection timeout. The consumer sends periodic heartbeats (heartbeat.interval.ms) to indicate its liveness to the broker. If no hearts are received by the broker for a group member within the session timeout, the broker will remove the consumer from the group and trigger a rebalance. The allowed range is configured with the **broker** configuration properties `group.min.session.timeout.ms` and `group.max.session.timeout.ms`. Also see `max.poll.interval.ms`. <br>*Type: integer*
 heartbeat.interval.ms                    |  C  | 1 .. 3600000    |          3000 | low        | Group session keepalive heartbeat interval. <br>*Type: integer*
-group.protocol.type                      |  C  |                 |      consumer | low        | Group protocol type for the `generic` group protocol. NOTE: Currently, the only supported group protocol type is `consumer`. <br>*Type: string*
+group.protocol.type                      |  C  |                 |      consumer | low        | Group protocol type for the `classic` group protocol. NOTE: Currently, the only supported group protocol type is `consumer`. <br>*Type: string*
+group.protocol                           |  C  | classic, consumer |       classic | high       | Group protocol to use. Use `classic` for the original protocol and `consumer` for the new protocol introduced in KIP-848. Available protocols: classic or consumer. Default is `classic`, but will change to `consumer` in next releases. <br>*Type: enum value*
+group.remote.assignor                    |  C  |                 |               | medium     | Server side assignor to use. Keep it null to make server select a suitable assignor for the group. Available assignors: uniform or range. Default is null <br>*Type: string*
 coordinator.query.interval.ms            |  C  | 1 .. 3600000    |        600000 | low        | How often to query for the current client group coordinator. If the currently assigned coordinator is down the configured query interval will be divided by ten to more quickly recover in case of coordinator reassignment. <br>*Type: integer*
 max.poll.interval.ms                     |  C  | 1 .. 86400000   |        300000 | high       | Maximum allowed time between calls to consume messages (e.g., rd_kafka_consumer_poll()) for high-level consumers. If this interval is exceeded the consumer is considered failed and the group will rebalance in order to reassign the partitions to another consumer group member. Warning: Offset commits may be not possible at this point. Note: It is recommended to set `enable.auto.offset.store=false` for long-time processing applications and then explicitly store offsets (using offsets_store()) *after* message processing, to make sure offsets are not auto-committed prior to processing has finished. The interval is checked two times per second. See KIP-62 for more information. <br>*Type: integer*
 enable.auto.commit                       |  C  | true, false     |          true | high       | Automatically and periodically commit offsets in the background. Note: setting this to false does not prevent the consumer from fetching previously committed start offsets. To circumvent this behaviour set specific start offsets per partition in the call to assign(). <br>*Type: boolean*

src/rdkafka.c

@@ -2186,7 +2186,7 @@ rd_kafka_t *rd_kafka_new(rd_kafka_type_t type,
         rd_kafka_resp_err_t ret_err = RD_KAFKA_RESP_ERR_NO_ERROR;
         int ret_errno               = 0;
         const char *conf_err;
-        rd_kafka_assignor_t *cooperative_assignor;
+        char *group_remote_assignor_override = NULL;
 #ifndef _WIN32
         sigset_t newset, oldset;
 #endif
@@ -2378,24 +2378,62 @@ rd_kafka_t *rd_kafka_new(rd_kafka_type_t type,
                 goto fail;
         }
 
-        /* Detect if chosen assignor is cooperative */
-        cooperative_assignor = rd_kafka_assignor_find(rk, "cooperative-sticky");
-        rk->rk_conf.partition_assignors_cooperative =
-            !rk->rk_conf.partition_assignors.rl_cnt ||
-            (cooperative_assignor && cooperative_assignor->rkas_enabled);
-
         if (!rk->rk_conf.group_remote_assignor) {
-                /* Default remote assignor to the chosen local one. */
-                if (rk->rk_conf.partition_assignors_cooperative) {
-                        rk->rk_conf.group_remote_assignor =
-                            rd_strdup("uniform");
-                } else {
-                        rd_kafka_assignor_t *range_assignor =
-                            rd_kafka_assignor_find(rk, "range");
-                        if (range_assignor && range_assignor->rkas_enabled)
+                rd_kafka_assignor_t *cooperative_assignor;
+
+                /* Detect if chosen assignor is cooperative
+                 * FIXME: remove this compatibility altogether
+                 * and apply the breaking changes that will be required
+                 * in next major version. */
+
+                cooperative_assignor =
+                    rd_kafka_assignor_find(rk, "cooperative-sticky");
+                rk->rk_conf.partition_assignors_cooperative =
+                    !rk->rk_conf.partition_assignors.rl_cnt ||
+                    (cooperative_assignor &&
+                     cooperative_assignor->rkas_enabled);
+
+                if (rk->rk_conf.group_protocol ==
+                    RD_KAFKA_GROUP_PROTOCOL_CONSUMER) {
+                        /* Default remote assignor to the chosen local one. */
+                        if (rk->rk_conf.partition_assignors_cooperative) {
+                                group_remote_assignor_override =
+                                    rd_strdup("uniform");
                                 rk->rk_conf.group_remote_assignor =
-                                    rd_strdup("range");
+                                    group_remote_assignor_override;
+                        } else {
+                                rd_kafka_assignor_t *range_assignor =
+                                    rd_kafka_assignor_find(rk, "range");
+                                if (range_assignor &&
+                                    range_assignor->rkas_enabled) {
+                                        rd_kafka_log(
+                                            rk, LOG_WARNING, "ASSIGNOR",
+                                            "\"range\" assignor is sticky "
+                                            "with group protocol CONSUMER");
+                                        group_remote_assignor_override =
+                                            rd_strdup("range");
+                                        rk->rk_conf.group_remote_assignor =
+                                            group_remote_assignor_override;
+                                } else {
+                                        rd_kafka_log(
+                                            rk, LOG_WARNING, "ASSIGNOR",
+                                            "roundrobin assignor isn't "
+                                            "available"
+                                            "with group protocol CONSUMER, "
+                                            "using the \"uniform\" one. "
+                                            "It's similar, "
+                                            "but it's also sticky");
+                                        group_remote_assignor_override =
+                                            rd_strdup("uniform");
+                                        rk->rk_conf.group_remote_assignor =
+                                            group_remote_assignor_override;
+                                }
+                        }
                 }
+        } else {
+                /* When users starts setting properties of the new protocol,
+                 * they can only use incremental_assign/unassign. */
+                rk->rk_conf.partition_assignors_cooperative = rd_true;
         }
 
         /* Create Mock cluster */
@@ -2667,6 +2705,8 @@ rd_kafka_t *rd_kafka_new(rd_kafka_type_t type,
          * that belong to rk_conf and thus needs to be cleaned up.
          * Legacy APIs, sigh.. */
         if (app_conf) {
+                if (group_remote_assignor_override)
+                        rd_free(group_remote_assignor_override);
                 rd_kafka_assignors_term(rk);
                 rd_kafka_interceptors_destroy(&rk->rk_conf);
                 memset(&rk->rk_conf, 0, sizeof(rk->rk_conf));

src/rdkafka_admin.c

@@ -6490,7 +6490,7 @@ rd_kafka_DeleteAclsResponse_parse(rd_kafka_op_t *rko_req,
                 result_response =
                     rd_kafka_DeleteAcls_result_response_new(error_code, errstr);
 
-                /* #maching_acls */
+                /* #matching_acls */
                 rd_kafka_buf_read_arraycnt(reply, &matching_acls_cnt, 100000);
                 for (j = 0; j < (int)matching_acls_cnt; j++) {
                         int16_t acl_error_code;

src/rdkafka_assignor.c

@@ -1279,7 +1279,7 @@ int verifyValidityAndBalance0(const char *func,
                                  * it means the assignment strategy failed to
                                  * properly balance the partitions. */
                                 if (!balanced &&
-                                    rd_kafka_topic_partition_list_find_topic(
+                                    rd_kafka_topic_partition_list_find_topic_by_name(
                                         otherPartitions, partition->topic)) {
                                         RD_UT_WARN(
                                             "Some %s partition(s) can be "