Skip to content
This repository has been archived by the owner on Aug 23, 2023. It is now read-only.

Kafka notifier: do not wait until timeout when there is no backlog to process, ie. no message in the partition queue #1315

Merged
merged 3 commits into from
May 23, 2019
Merged

Kafka notifier: do not wait until timeout when there is no backlog to process, ie. no message in the partition queue #1315

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a1cf12d
Kafka notifier: do not wait until timeout when there is no backlog to…
fkaleo May 17, 2019
c451e62
Rationalised NotifierKafka's backlog processing:
fkaleo May 17, 2019
a211433
Removed old comment that does not apply anylonger
fkaleo May 18, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion mdata/notifierKafka/notifierKafka.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,7 +123,16 @@ func (c *NotifierKafka) consumePartition(topic string, partition int32, currentO

messages := pc.Messages()
ticker := time.NewTicker(5 * time.Second)
startingUp := true

var startingUp bool
if bootTimeOffsets[partition] == 0 {
woodsaj marked this conversation as resolved.
Show resolved Hide resolved
// no message in the partition queue; no backlog to process
startingUp = false
processBacklog.Done()
} else {
startingUp = true
}

// the bootTimeOffset is the next available offset. There may not be a message with that
// offset yet, so we subtract 1 to get the highest offset that we can fetch.
bootTimeOffset := bootTimeOffsets[partition] - 1
Expand Down