initial time in stats

kafka/confluent/README.md

@@ -3,7 +3,7 @@ This a wrapper around the confluent-kafka-go library.
 To use library
 - Create processor or view with confluent consumer, eg, `goka.WithConsumerBuilder(confluent.NewConsumerBuilder(1000))`
 - Install `librdkafka` in the compilation environment
-- Compile the go binary with `-tags confluent -tags static`
+- Compile the go binary with `-tags "confluent static"`
 - Install `libssl1.0.0` `libsasl2-2` (or equivalent) in the execution environment
 
 Note that this is experimental, not well tested and features are missing (in particular `auto.commit` is set to true).

partition.go

@@ -379,7 +379,7 @@ func (p *partition) fetchStats() *PartitionStats {
 	case p.requestStats <- true:
 	case <-p.dying:
 		// if closing, return empty stats
-		return newPartitionStats()
+		return newPartitionStats().init(p.lastStats, p.offset, p.hwm)
 	case <-timer.C:
 		return p.lastStats
 	}
@@ -389,6 +389,6 @@ func (p *partition) fetchStats() *PartitionStats {
 		return s
 	case <-p.dying:
 		// if closing, return empty stats
-		return newPartitionStats()
+		return newPartitionStats().init(p.lastStats, p.offset, p.hwm)
 	}
 }

stats.go

@@ -42,6 +42,7 @@ type PartitionStats struct {
 
 func newPartitionStats() *PartitionStats {
 	return &PartitionStats{
+		Now:    time.Now(),
 		Input:  make(map[string]InputStats),
 		Output: make(map[string]OutputStats),
 	}