PR #4970: Code and tests fixes to make the full test suite pass

Author: emasab

.semaphore/run-all-tests.yml

@@ -0,0 +1,77 @@
+version: v1.0
+name: run-all-tests
+
+agent:
+  machine:
+    type: s1-prod-ubuntu22-04-amd64-1
+
+execution_time_limit:
+  hours: 3
+
+global_job_config:
+  prologue:
+    commands:
+      - checkout
+      - '[[ -z "$GIT_REF" ]] || git checkout $GIT_REF'
+      - wget -O rapidjson-dev.deb https://launchpad.net/ubuntu/+archive/primary/+files/rapidjson-dev_1.1.0+dfsg2-3_all.deb
+      - sudo dpkg -i rapidjson-dev.deb
+      - sudo apt update
+      - sudo apt remove -y needrestart
+      - sudo apt install -y valgrind
+      - python3 -m pip install -U pip
+      - python3 -m pip -V
+      - (cd tests && python3 -m pip install -r requirements.txt)
+      - ./configure --install-deps
+      - make -j all
+      - make -j -C tests build
+      - sem-version java 17
+
+blocks:
+  - name: "Run all tests (x86_64)"
+    dependencies: []
+    task:
+      agent:
+        machine:
+          type: s1-prod-ubuntu22-04-amd64-2
+      prologue:
+        commands:
+          - if [[ "$TEST_ARCHES" != *"x86_64"* ]]; then exit 0; fi
+      jobs:
+        - name: "PLAINTEXT cluster (x86_64)"
+          env_vars:
+            - name: TEST_SSL
+              value: "False"
+          commands:
+            - if [[ "$TEST_TYPE" != *"plaintext"* ]]; then exit 0; fi
+            - ./tests/run-all-tests.sh 
+        - name: "SSL cluster (x86_64)"
+          env_vars:
+            - name: TEST_SSL
+              value: "True"
+          commands:
+            - if [[ "$TEST_TYPE" != *"ssl"* ]]; then exit 0; fi
+            - ./tests/run-all-tests.sh
+  - name: "Run all tests (aarch64)"
+    dependencies: []
+    task:
+      agent:
+        machine:
+          type: s1-prod-ubuntu22-04-arm64-2
+      prologue:
+        commands:
+          - if [[ "$TEST_ARCHES" != *"aarch64"* ]]; then exit 0; fi
+      jobs:
+        - name: "PLAINTEXT cluster (aarch64)"
+          env_vars:
+            - name: TEST_SSL
+              value: "False"
+          commands:
+            - if [[ "$TEST_TYPE" != *"plaintext"* ]]; then exit 0; fi
+            - ./tests/run-all-tests.sh
+        - name: "SSL cluster (aarch64)"
+          env_vars:
+            - name: TEST_SSL
+              value: "True"
+          commands:
+            - if [[ "$TEST_TYPE" != *"ssl"* ]]; then exit 0; fi
+            - ./tests/run-all-tests.sh

.semaphore/semaphore.yml

@@ -384,3 +384,25 @@ blocks:
             # Upload all packages to project artifact store
             - artifact push project packages --destination librdkafka-packages-${SEMAPHORE_GIT_TAG_NAME}-${SEMAPHORE_WORKFLOW_ID}
             - echo Thank you
+
+promotions:
+  - name: Run all tests on master commits
+    pipeline_file: run-all-tests.yml
+    parameters:
+      env_vars:
+        - required: true
+          name: TEST_KAFKA_GIT_REF
+          default_value: 3.8.0
+        - required: true
+          name: TEST_TYPE
+          default_value: plaintext,ssl
+        - required: true
+          name: TEST_ARCHES
+          default_value: x86_64,aarch64
+        - required: true
+          name: TEST_PARALLEL
+          default_value: "1"
+    auto_promote_on:
+      - result: passed
+        branch:
+        - "master"

CHANGELOG.md

@@ -1,8 +1,36 @@
 # librdkafka v2.9.0
 
+librdkafka v2.9.0 is a feature release:
+
  * Identify brokers only by broker id (#4557, @mfleming)
  * Remove unavailable brokers and their thread (#4557, @mfleming)
-
+ * Fix for librdkafka yielding before timeouts had been reached (#)
+ * Removed a 500ms latency when a consumer partition switches to a different
+   leader (#)
+ * The mock cluster implementation removes brokers from Metadata response
+   when they're not available, this simulates better the actual behavior of
+   a cluster that is using KRaft (#).
+ * Doesn't remove topics from cache on temporary Metadata errors but only
+   on metadata cache expiry (#).
+ * Doesn't mark the topic as unknown if it had been marked as existent earlier
+   and `topic.metadata.propagation.max.ms` hasn't passed still (#).
+ * Doesn't update partition leaders if the topic in metadata
+   response has errors (#).
+ * Only topic authorization errors in a metadata response are considered
+   permanent and are returned to the user (#).
+ * The function `rd_kafka_offsets_for_times` refreshes leader information
+   if the error requires it, allowing it to succeed on
+   subsequent manual retries (#).
+ * Deprecated `api.version.request`, `api.version.fallback.ms` and
+   `broker.version.fallback` configuration properties (#).
+ * When consumer is closed before destroying the client, the operations queue
+   isn't purged anymore as it contains operations
+   unrelated to the consumer group (#).
+ * When making multiple changes to the consumer subscription in a short time,
+   no unknown topic error is returned for topics that are in the new subscription but weren't in previous one (#).
+ * Fix for the case where a metadata refresh enqueued on an unreachable broker
+   prevents refreshing the controller or the coordinator until that broker
+   becomes reachable again (#).
 
 ## Fixes
 
@@ -20,6 +48,74 @@
    temporarily or permanently so we always remove it and it'll be added back when
    it becomes available again.
    Happens since 1.x (#4557, @mfleming).
+ * Issues: #
+   librdkafka code using `cnd_timedwait` was yielding before a timeout occurred
+   without the condition being fulfilled because of spurious wake-ups.
+   Solved by verifying with a monotonic clock that the expected point in time
+   was reached and calling the function again if needed.
+   Happens since 1.x (#).
+ * Issues: #
+   Doesn't remove topics from cache on temporary Metadata errors but only
+   on metadata cache expiry. It allows the client to continue working
+   in case of temporary problems to the Kafka metadata plane.
+   Happens since 1.x (#).
+ * Issues: #
+   Doesn't mark the topic as unknown if it had been marked as existent earlier
+   and `topic.metadata.propagation.max.ms` hasn't passed still. It achieves
+   this property expected effect even if a different broker had
+   previously reported the topic as existent.
+   Happens since 1.x (#).
+ * Issues: #
+   Doesn't update partition leaders if the topic in metadata
+   response has errors. It's in line with what Java client does and allows
+   to avoid segmentation faults for unknown partitions.
+   Happens since 1.x (#).
+ * Issues: #
+   Only topic authorization errors in a metadata response are considered
+   permanent and are returned to the user. It's in line with what Java client
+   does and avoids returning to the user an error that wasn't meant to be
+   permanent.
+   Happens since 1.x (#).
+ * Issues: #
+   Fix for the case where a metadata refresh enqueued on an unreachable broker
+   prevents refreshing the controller or the coordinator until that broker
+   becomes reachable again. Given the request continues to be retried on that
+   broker, the counter for refreshing complete broker metadata doesn't reach
+   zero and prevents the client from obtaining the new controller or group or transactional coordinator.
+   It causes a series of debug messages like:
+   "Skipping metadata request: ... full request already in-transit", until
+   the broker the request is enqueued on is up again.
+   Solved by not retrying these kinds of metadata requests.
+   Happens since 1.x (#).
+
+### Consumer fixes
+
+ * Issues: #
+   When switching to a different leader a consumer could wait 500ms 
+   (`fetch.error.backoff.ms`) before starting to fetch again. The fetch backoff wasn't reset when joining the new broker.
+   Solved by resetting it, given it's not needed to backoff
+   the first fetch on a different node. This way faster leader switches are
+   possible.
+   Happens since 1.x (#).
+ * Issues: #
+   The function `rd_kafka_offsets_for_times` refreshes leader information
+   if the error requires it, allowing it to succeed on
+   subsequent manual retries. Similar to the fix done in 2.3.0 in
+   `rd_kafka_query_watermark_offsets`. Additionally, the partition
+   current leader epoch is taken from metadata cache instead of
+   from passed partitions.
+   Happens since 1.x (#).
+ * Issues: #
+   When consumer is closed before destroying the client, the operations queue
+   isn't purged anymore as it contains operations
+   unrelated to the consumer group.
+   Happens since 1.x (#).
+ * Issues: #
+   When making multiple changes to the consumer subscription in a short time,
+   no unknown topic error is returned for topics that are in the new subscription
+   but weren't in previous one. This was due to the metadata request relative
+   to previous subscription.
+   Happens since 1.x (#).
 
 
 

CONFIGURATION.md

@@ -54,10 +54,10 @@ resolve_cb                               |  *  |                 |
 opaque                                   |  *  |                 |               | low        | Application opaque (set with rd_kafka_conf_set_opaque()) <br>*Type: see dedicated API*
 default_topic_conf                       |  *  |                 |               | low        | Default topic configuration for automatically subscribed topics <br>*Type: see dedicated API*
 internal.termination.signal              |  *  | 0 .. 128        |             0 | low        | Signal that librdkafka will use to quickly terminate on rd_kafka_destroy(). If this signal is not set then there will be a delay before rd_kafka_wait_destroyed() returns true as internal threads are timing out their system calls. If this signal is set however the delay will be minimal. The application should mask this signal as an internal signal handler is installed. <br>*Type: integer*
-api.version.request                      |  *  | true, false     |          true | high       | Request broker's supported API versions to adjust functionality to available protocol features. If set to false, or the ApiVersionRequest fails, the fallback version `broker.version.fallback` will be used. **NOTE**: Depends on broker version >=0.10.0. If the request is not supported by (an older) broker the `broker.version.fallback` fallback is used. <br>*Type: boolean*
+api.version.request                      |  *  | true, false     |          true | high       | **DEPRECATED** **Post-deprecation actions: remove this configuration property, brokers < 0.10.0 won't be supported anymore in librdkafka 3.x.** Request broker's supported API versions to adjust functionality to available protocol features. If set to false, or the ApiVersionRequest fails, the fallback version `broker.version.fallback` will be used. **NOTE**: Depends on broker version >=0.10.0. If the request is not supported by (an older) broker the `broker.version.fallback` fallback is used. <br>*Type: boolean*
 api.version.request.timeout.ms           |  *  | 1 .. 300000     |         10000 | low        | Timeout for broker API version requests. <br>*Type: integer*
-api.version.fallback.ms                  |  *  | 0 .. 604800000  |             0 | medium     | Dictates how long the `broker.version.fallback` fallback is used in the case the ApiVersionRequest fails. **NOTE**: The ApiVersionRequest is only issued when a new connection to the broker is made (such as after an upgrade). <br>*Type: integer*
-broker.version.fallback                  |  *  |                 |        0.10.0 | medium     | Older broker versions (before 0.10.0) provide no way for a client to query for supported protocol features (ApiVersionRequest, see `api.version.request`) making it impossible for the client to know what features it may use. As a workaround a user may set this property to the expected broker version and the client will automatically adjust its feature set accordingly if the ApiVersionRequest fails (or is disabled). The fallback broker version will be used for `api.version.fallback.ms`. Valid values are: 0.9.0, 0.8.2, 0.8.1, 0.8.0. Any other value >= 0.10, such as 0.10.2.1, enables ApiVersionRequests. <br>*Type: string*
+api.version.fallback.ms                  |  *  | 0 .. 604800000  |             0 | medium     | **DEPRECATED** **Post-deprecation actions: remove this configuration property, brokers < 0.10.0 won't be supported anymore in librdkafka 3.x.** Dictates how long the `broker.version.fallback` fallback is used in the case the ApiVersionRequest fails. **NOTE**: The ApiVersionRequest is only issued when a new connection to the broker is made (such as after an upgrade). <br>*Type: integer*
+broker.version.fallback                  |  *  |                 |        0.10.0 | medium     | **DEPRECATED** **Post-deprecation actions: remove this configuration property, brokers < 0.10.0 won't be supported anymore in librdkafka 3.x.** Older broker versions (before 0.10.0) provide no way for a client to query for supported protocol features (ApiVersionRequest, see `api.version.request`) making it impossible for the client to know what features it may use. As a workaround a user may set this property to the expected broker version and the client will automatically adjust its feature set accordingly if the ApiVersionRequest fails (or is disabled). The fallback broker version will be used for `api.version.fallback.ms`. Valid values are: 0.9.0, 0.8.2, 0.8.1, 0.8.0. Any other value >= 0.10, such as 0.10.2.1, enables ApiVersionRequests. <br>*Type: string*
 allow.auto.create.topics                 |  *  | true, false     |         false | low        | Allow automatic topic creation on the broker when subscribing to or assigning non-existent topics. The broker must also be configured with `auto.create.topics.enable=true` for this configuration to take effect. Note: the default value (true) for the producer is different from the default value (false) for the consumer. Further, the consumer default value is different from the Java consumer (true), and this property is not supported by the Java producer. Requires broker version >= 0.11.0.0, for older broker versions only the broker configuration applies. <br>*Type: boolean*
 security.protocol                        |  *  | plaintext, ssl, sasl_plaintext, sasl_ssl |     plaintext | high       | Protocol used to communicate with brokers. <br>*Type: enum value*
 ssl.cipher.suites                        |  *  |                 |               | low        | A cipher suite is a named combination of authentication, encryption, MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol. See manual page for `ciphers(1)` and `SSL_CTX_set_cipher_list(3). <br>*Type: string*