Add client function for fetching topic replica count

[lairen]

For tooling / automation scripts, it would be super helpful to expose a method in the client object for querying a topic's replica count, which is possible from the cluster.partitions_for function.

ruby-kafka/lib/kafka/cluster.rb

Line 180 in 1b2ed7c

def partitions_for(topic)

Another option could be to write a client function that returns the full partition array instead of just the count then I can grab the replica count from my code.

ruby-kafka/lib/kafka/client.rb

Line 694 in 1b2ed7c

@cluster.partitions_for(topic).count

Let me know what you think or if there's a better place for this.

[dasch]

Thanks!

[theturtle32]

Nice convenience function. Keep in mind, though, that while I imagine it would be very uncommon, it's technically possible for not all partitions to have the same replica count.

[lairen]

@theturtle32 I've found that to be true during partition reassignment operations specifically, so that is one caveat to this function. Even though that is uncommon in my experience, it might be worth noting that in the documentation.

It'd be great if there was a more consistent way to get a topic's replication factor from Kafka, but I wasn't able to find one.

[lairen]

Just did some quick digging. In earlier versions of Kafka, the kafka-topics command line tool evaluates the topic's replication factor by counting the replicas on the first partition object. However, it looks like in Kafka v2.5, the tool checks to see if a partition reassignment is in progress when evaluating the replication factor (e.g. TopicCommand.scala ).

I see now that the Kafka AdminClient has a ListPartitionReassignments API that could be implemented in this library in a similar way to make this function more reliable 🤔

ListPartitionReassignments Response (Version: 0) => throttle_time_ms error_code error_message [topics] TAG_BUFFER 
  throttle_time_ms => INT32
  error_code => INT16
  error_message => COMPACT_NULLABLE_STRING
  topics => name [partitions] TAG_BUFFER 
    name => COMPACT_STRING
    partitions => partition_index [replicas] [adding_replicas] [removing_replicas] TAG_BUFFER 
      partition_index => INT32
      replicas => INT32
      adding_replicas => INT32
      removing_replicas => INT32