ListOffsets KIP : 396

CHANGELOG.md

@@ -13,6 +13,8 @@ librdkafka v2.3.0 is a feature release:
  * [KIP-580](https://cwiki.apache.org/confluence/display/KAFKA/KIP-580%3A+Exponential+Backoff+for+Kafka+Clients): Added Exponential Backoff mechanism for
    retriable requests with `retry.backoff.ms` as minimum backoff and `retry.backoff.max.ms` as the
    maximum backoff, with 20% jitter (#4422).
+ * [KIP-396](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=97551484): completed the implementation with
+   the addition of ListOffsets (#4225).
  * Fixed ListConsumerGroupOffsets not fetching offsets for all the topics in a group with Apache Kafka version below 2.4.0.
  * Add missing destroy that leads to leaking partition structure memory when there
    are partition leader changes and a stale leader epoch is received (#4429).
@@ -32,6 +34,8 @@ librdkafka v2.3.0 is a feature release:
    consume_cb (#4431).
  * Fix for idempotent producer fatal errors, triggered after a possibly persisted message state (#4438).
  * Fix `rd_kafka_query_watermark_offsets` continuing beyond timeout expiry (#4460).
+ * Fix `rd_kafka_query_watermark_offsets` not refreshing the partition leader
+   after a leader change and subsequent `NOT_LEADER_OR_FOLLOWER` error (#4225).
 
 
 ## Upgrade considerations
@@ -54,9 +58,9 @@ librdkafka v2.3.0 is a feature release:
  * An assertion failed with insufficient buffer size when allocating
    rack information on 32bit architectures.
    Solved by aligning all allocations to the maximum allowed word size (#4449).
- * The timeout for `rd_kafka_query_watermark_offsets` was not checked after
+ * The timeout for `rd_kafka_query_watermark_offsets` was not enforced after
    making the necessary ListOffsets requests, and thus, it never timed out in
-   case of broker/network issues. Fixed by checking timeout expiry (#4460).
+   case of broker/network issues. Fixed by setting an absolute timeout (#4460).
 
 ### Idempotent producer fixes
 
@@ -93,6 +97,10 @@ librdkafka v2.3.0 is a feature release:
     consumer messages, while the method to service the queue internally also
     services the queue forwarded to from `rk_rep`, which is `rkcg_q`.
     Solved by moving the `max.poll.interval.ms` check into `rd_kafka_q_serve` (#4431).
+  * After a leader change a `rd_kafka_query_watermark_offsets` call would continue
+    trying to call ListOffsets on the old leader, if the topic wasn't included in
+    the subscription set, so it started querying the new leader only after
+    `topic.metadata.refresh.interval.ms` (#4225).
 
 
 

INTRODUCTION.md

@@ -319,7 +319,7 @@ error code set.
 
 The application should typically not attempt to retry producing the message
 on failure, but instead configure librdkafka to perform these retries
-using the `retries`, `retry.backoff.ms` and `retry.backoff.max.ms` 
+using the `retries`, `retry.backoff.ms` and `retry.backoff.max.ms`
 configuration properties.
 
 
@@ -1923,7 +1923,7 @@ The [Apache Kafka Implementation Proposals (KIPs)](https://cwiki.apache.org/conf
 | KIP-389 - Consumer group max size                                        | 2.2.0                       | Supported (error is propagated to application, but the consumer does not raise a fatal error) |
 | KIP-392 - Allow consumers to fetch from closest replica                  | 2.4.0                       | Supported                                                                                     |
 | KIP-394 - Consumer: require member.id in JoinGroupRequest                | 2.2.0                       | Supported                                                                                     |
-| KIP-396 - AdminAPI: commit/list offsets                                  | 2.4.0                       | Partially supported (remaining APIs available outside Admin client)                           |
+| KIP-396 - AdminAPI: commit/list offsets                                  | 2.4.0                       | Supported                                                                                     |
 | KIP-412 - AdminAPI: adjust log levels                                    | 2.4.0                       | Not supported                                                                                 |
 | KIP-421 - Variables in client config files                               | 2.3.0                       | Not applicable (librdkafka, et.al, does not provide a config file interface, and shouldn't)   |
 | KIP-429 - Consumer: incremental rebalance protocol                       | 2.4.0                       | Supported                                                                                     |
@@ -1976,7 +1976,7 @@ release of librdkafka.
 | ------- | ------------------------------| ----------- | ----------------------- |
 | 0       | Produce                       | 9           | 7                       |
 | 1       | Fetch                         | 15          | 11                      |
-| 2       | ListOffsets                   | 8           | 5                       |
+| 2       | ListOffsets                   | 8           | 7                       |
 | 3       | Metadata                      | 12          | 12                      |
 | 8       | OffsetCommit                  | 8           | 7                       |
 | 9       | OffsetFetch                   | 8           | 7                       |

examples/.gitignore

@@ -20,4 +20,5 @@ describe_cluster
 list_consumer_group_offsets
 alter_consumer_group_offsets
 incremental_alter_configs
-user_scram
+user_scram
+list_offsets

examples/CMakeLists.txt

@@ -59,6 +59,9 @@ target_link_libraries(describe_topics PUBLIC rdkafka)
 add_executable(describe_cluster describe_cluster.c ${win32_sources})
 target_link_libraries(describe_cluster PUBLIC rdkafka)
 
+add_executable(list_offsets list_offsets.c ${win32_sources})
+target_link_libraries(list_offsets PUBLIC rdkafka)
+
 # The targets below has Unix include dirs and do not compile on Windows.
 if(NOT WIN32)
     add_executable(rdkafka_example rdkafka_example.c)
@@ -69,4 +72,5 @@ if(NOT WIN32)
 
     add_executable(kafkatest_verifiable_client kafkatest_verifiable_client.cpp)
     target_link_libraries(kafkatest_verifiable_client PUBLIC rdkafka++)
+
 endif(NOT WIN32)

examples/Makefile

@@ -12,6 +12,7 @@ EXAMPLES ?= rdkafka_example rdkafka_performance rdkafka_example_cpp \
 	alter_consumer_group_offsets \
 	incremental_alter_configs \
 	user_scram \
+	list_offsets \
 	misc
 
 all: $(EXAMPLES)
@@ -148,6 +149,10 @@ user_scram: ../src/librdkafka.a user_scram.c
 	$(CC) $(CPPFLAGS) $(CFLAGS) $@.c -o $@ $(LDFLAGS) \
 		../src/librdkafka.a $(LIBS)
 
+list_offsets: ../src/librdkafka.a list_offsets.c
+	$(CC) $(CPPFLAGS) $(CFLAGS) $@.c -o $@ $(LDFLAGS) \
+		../src/librdkafka.a $(LIBS)
+
 misc: ../src/librdkafka.a misc.c
 	$(CC) $(CPPFLAGS) $(CFLAGS) $@.c -o $@ $(LDFLAGS) \
 		../src/librdkafka.a $(LIBS)