Merge pull request #1328 from grafana/kafka_notifier_timeout_oldest

Dieterbe · web-flow · commit 9b52b1a5ca44 · 2019-05-27T12:57:14.000+03:00
NotifierKafka: avoid timeout case when there are no persist messages and using OffsetOldest or offsetDuration
diff --git a/mdata/notifierKafka/notifierKafka.go b/mdata/notifierKafka/notifierKafka.go
@@ -82,6 +82,10 @@ func (c *NotifierKafka) start() {
 		if err != nil {
 			log.Fatalf("kafka-cluster: failed to get offset %d: %s", offsetTime, err)
 		}
+		if startOffset < 0 {
+			// happens when OffsetOldest or an offsetDuration was used and there is no message in the partition
+			startOffset = 0
+		}
 		processBacklog.Add(1)
 		go c.consumePartition(topic, partition, startOffset, processBacklog)
 	}

Original file line number	Diff line number	Diff line change
`@@ -82,6 +82,10 @@ func (c *NotifierKafka) start() {`
`82`	`82`	`if err != nil {`
`83`	`83`	`log.Fatalf("kafka-cluster: failed to get offset %d: %s", offsetTime, err)`
`84`	`84`	`}`
	`85`	`+ if startOffset < 0 {`
	`86`	`+ // happens when OffsetOldest or an offsetDuration was used and there is no message in the partition`
	`87`	`+ startOffset = 0`
	`88`	`+ }`
`85`	`89`	`processBacklog.Add(1)`
`86`	`90`	`go c.consumePartition(topic, partition, startOffset, processBacklog)`
`87`	`91`	`}`