Make topic with compact.

annymsMthd · annymsMthd · commit c9fb0af22f2c · 2019-12-17T20:49:09.000-08:00
(+semver: fix)

Signed-off-by: Joshua Benjamin &lt;joshuabnjmn@gmail.com&gt;
diff --git a/internal/storage/kafka.go b/internal/storage/kafka.go
@@ -65,9 +65,13 @@ func newKafkaConsumer(broker string, replicas int16, storer SchemaStorer, topic
 	}
 	defer admin.Close()
 
+	compact := "compact"
 	topicDetail := &sarama.TopicDetail{
 		NumPartitions:     1,
 		ReplicationFactor: replicas,
+		ConfigEntries: map[string]*string{
+			"cleanup.policy": &compact,
+		},
 	}
 	err = admin.CreateTopic(topic, topicDetail, false)
 	if err != nil && !strings.Contains(err.Error(), "already exists") {

Original file line number	Diff line number	Diff line change
`@@ -65,9 +65,13 @@ func newKafkaConsumer(broker string, replicas int16, storer SchemaStorer, topic`
`65`	`65`	`}`
`66`	`66`	`defer admin.Close()`
`67`	`67`
	`68`	`+ compact := "compact"`
`68`	`69`	`topicDetail := &sarama.TopicDetail{`
`69`	`70`	`NumPartitions: 1,`
`70`	`71`	`ReplicationFactor: replicas,`
	`72`	`+ ConfigEntries: map[string]*string{`
	`73`	`+ "cleanup.policy": &compact,`
	`74`	`+ },`
`71`	`75`	`}`
`72`	`76`	`err = admin.CreateTopic(topic, topicDetail, false)`
`73`	`77`	`if err != nil && !strings.Contains(err.Error(), "already exists") {`