Add options to include topic metadata and offsets in list

commands/list/topics.go

@@ -18,11 +18,13 @@ import (
 )
 
 type topics struct {
-	kafkaParams  *commands.KafkaParameters
-	globalParams *commands.GlobalParameters
-	topicFilter  *regexp.Regexp
-	format       string
-	style        string
+	kafkaParams    *commands.KafkaParameters
+	globalParams   *commands.GlobalParameters
+	topicFilter    *regexp.Regexp
+	loadConfigs    bool
+	includeOffsets bool
+	format         string
+	style          string
 }
 
 func addTopicsSubCommand(parent *kingpin.CmdClause, global *commands.GlobalParameters, kafkaParams *commands.KafkaParameters) {
@@ -34,6 +36,12 @@ func addTopicsSubCommand(parent *kingpin.CmdClause, global *commands.GlobalParam
 	c.Flag("topic-filter", "An optional regular expression to filter the topics by.").
 		Short('t').
 		RegexpVar(&cmd.topicFilter)
+	c.Flag("load-config", "Loads the topic's configurations from the server.").
+		NoEnvar().
+		Short('c').BoolVar(&cmd.loadConfigs)
+	c.Flag("include-offsets", "Queries the server to read the latest available offset of each partition.").
+		NoEnvar().
+		Short('o').BoolVar(&cmd.includeOffsets)
 	commands.AddFormatFlag(c, &cmd.format, &cmd.style)
 }
 
@@ -49,7 +57,7 @@ func (c *topics) run(_ *kingpin.ParseContext) error {
 		cancel()
 	}()
 
-	topics, err := manager.GetTopics(ctx, c.topicFilter)
+	topics, err := manager.GetTopics(ctx, c.topicFilter, c.loadConfigs, c.includeOffsets)
 	if err != nil {
 		return err
 	}

kafka/manager.go

@@ -191,7 +191,7 @@ func (m *Manager) DescribeCluster(ctx context.Context, includeConfig bool) (*Clu
 }
 
 // GetTopics returns a list of all the topics on the server.
-func (m *Manager) GetTopics(ctx context.Context, filter *regexp.Regexp) ([]Topic, error) {
+func (m *Manager) GetTopics(ctx context.Context, filter *regexp.Regexp, includeConfig, includeOffsets bool) ([]Topic, error) {
 	m.Log(internal.Verbose, "Retrieving topic list from the server")
 	topics, err := m.admin.ListTopics()
 	if err != nil {
@@ -209,11 +209,19 @@ func (m *Manager) GetTopics(ctx context.Context, filter *regexp.Regexp) ([]Topic
 				m.Logf(internal.SuperVerbose, "Filtering out %s topic", topic)
 				continue
 			}
-			result = append(result, Topic{
+			t := Topic{
 				Name:               topic,
 				NumberOfPartitions: details.NumPartitions,
 				ReplicationFactor:  details.ReplicationFactor,
-			})
+			}
+			if includeConfig {
+				meta, err := m.DescribeTopic(ctx, topic, includeConfig, includeOffsets)
+				if err == nil {
+					t.Metadata = meta
+
+				}
+			}
+			result = append(result, t)
 		}
 	}
 	return result, nil

kafka/topic.go

@@ -10,6 +10,8 @@ type Topic struct {
 	NumberOfPartitions int32 `json:"number_of_partitions"`
 	// ReplicationFactor replication factor.
 	ReplicationFactor int16 `json:"replication_factor"`
+	// Topic metadata.
+	Metadata           *TopicMetadata `json:"metadata"`
 }
 
 // TopicsByName sorts the topic list by name.