[DP-1762] - topicctl get actions for under replicated and offline partitions

[ssingudasu]

Description

This PR is to add new actions for topicctl get

Expectations:

• topicctl get under-replicated-partitions (optional --topics flag)
• topicctl get offline-partitions (optional --topics flag)
• Fix the debug PersistentPreRunE: getPreRun in topicctl get since topicctl get is overriding the root cmd PersistentPreRunE: preRun

under replicated partitions

• Kafka cluster is in under replicated state if the number of ISR are less than the Replicas available for the partition
• topicctl already has readily available metadata call with filtered information. We will leverage topicctl admin pkg for under-replicated-partitions

offline partitions

• topicctl has admin.GetTopics which performs kafka Metadata call, there is a lot of formatting and missing information in admin.GetTopics. Hence we will perform the kafka-go client.Metadata to ensure we capture all the offline partition information.
• We leverage kafka-go metadata for offline-partitions

kafka-go metadata call for offline partitions gives the Leader partition information as below:

kafka "github.com/segmentio/kafka-go"

partition:

{
    Topic: topic_name
    ID: partition_id
    Leader: kafka.Broker (refer below)
    Replicas: [][Broker
    Isr:  [][Broker
    Error: kafka.Error
}

partition.Leader Broker response:

kafka.Broker{
        Host: ""
    Port: 0
    ID: 0
    Rack: ""
}

Unlike sarama which gives the offline leader broker id as -1, kafka-go gives the offline broker id as 0

Hence we will check the below error code: ListenerNotFound along with broker.Host as "" and broker.Port as 0 for partitions information in the metadata response

• error message: "there is no listener on the leader broker that matches the listener on which metadata request was processed"
• kafka.Error reference: https://github.com/segmentio/kafka-go/blob/main/error.go#L87

Expectation: if kafka.Error is ListenerNotFound and there is no broker or port found for leader partition, there by confirming that the partition is offline. We will display the ISR.ID and Repicas.ID with information from the kafka-go metadata call (i.e: each offline partition - Isr.ID and Replicas.ID as 0)

[petedannemann]

Could we document these new commands in the README with the other get commands?

[ssingudasu]

Closing this PR as we have decided to implement the same in

# topicctl get partitions