<commit_before>package notifierKafka

import (
	"bytes"
	"encoding/binary"
	"encoding/json"
	"sync"
	"time"

	"github.com/raintank/schema"

	"github.com/Shopify/sarama"
	"github.com/grafana/metrictank/mdata"
	"github.com/grafana/metrictank/util"
	log "github.com/sirupsen/logrus"
)

type NotifierKafka struct {
	instance string
	in       chan mdata.SavedChunk
	buf      []mdata.SavedChunk
	wg       sync.WaitGroup
	bPool    *util.BufferPool
	handler  mdata.NotifierHandler
	client   sarama.Client
	consumer sarama.Consumer
	producer sarama.SyncProducer
	StopChan chan int

	// signal to PartitionConsumers to shutdown
	stopConsuming chan struct{}
}

func New(instance string, handler mdata.NotifierHandler) *NotifierKafka {
	client, err := sarama.NewClient(brokers, config)
	if err != nil {
		log.Fatalf("kafka-cluster: failed to start client: %s", err)
	}
	consumer, err := sarama.NewConsumerFromClient(client)
	if err != nil {
		log.Fatalf("kafka-cluster: failed to initialize consumer: %s", err)
	}
	log.Info("kafka-cluster: consumer initialized without error")

	producer, err := sarama.NewSyncProducerFromClient(client)
	if err != nil {
		log.Fatalf("kafka-cluster: failed to initialize producer: %s", err)
	}

	c := NotifierKafka{
		instance: instance,
		in:       make(chan mdata.SavedChunk),
		bPool:    util.NewBufferPool(),
		handler:  handler,
		client:   client,
		consumer: consumer,
		producer: producer,

		StopChan:      make(chan int),
		stopConsuming: make(chan struct{}),
	}
	c.start()
	go c.produce()

	return &c
}

func (c *NotifierKafka) start() {
	pre := time.Now()
	processBacklog := new(sync.WaitGroup)
	for _, partition := range partitions {
		var offsetTime int64
		switch offsetStr {
		case "oldest":
			offsetTime = sarama.OffsetOldest
		case "newest":
			offsetTime = sarama.OffsetNewest
		default:
			offsetTime = time.Now().Add(-1*offsetDuration).UnixNano() / int64(time.Millisecond)
		}
		startOffset, err := c.client.GetOffset(topic, partition, offsetTime)
		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)
	}
	// wait for our backlog to be processed before returning.  This will block metrictank from consuming metrics until
	// we have processed old metricPersist messages. The end result is that we wont overwrite chunks in cassandra that
	// have already been previously written.
	// We don't wait more than backlogProcessTimeout for the backlog to be processed.
	log.Info("kafka-cluster: waiting for metricPersist backlog to be processed.")
	backlogProcessed := make(chan struct{}, 1)
	go func() {
		processBacklog.Wait()
		backlogProcessed <- struct{}{}
	}()<commit_msg>fix how the kafka offsets get defined in the notifier<commit_after>package notifierKafka

import (
	"bytes"
	"encoding/binary"
	"encoding/json"
	"sync"
	"time"

	"github.com/raintank/schema"

	"github.com/Shopify/sarama"
	"github.com/grafana/metrictank/mdata"
	"github.com/grafana/metrictank/util"
	log "github.com/sirupsen/logrus"
)

type NotifierKafka struct {
	instance string
	in       chan mdata.SavedChunk
	buf      []mdata.SavedChunk
	wg       sync.WaitGroup
	bPool    *util.BufferPool
	handler  mdata.NotifierHandler
	client   sarama.Client
	consumer sarama.Consumer
	producer sarama.SyncProducer
	StopChan chan int

	// signal to PartitionConsumers to shutdown
	stopConsuming chan struct{}
}

func New(instance string, handler mdata.NotifierHandler) *NotifierKafka {
	client, err := sarama.NewClient(brokers, config)
	if err != nil {
		log.Fatalf("kafka-cluster: failed to start client: %s", err)
	}
	consumer, err := sarama.NewConsumerFromClient(client)
	if err != nil {
		log.Fatalf("kafka-cluster: failed to initialize consumer: %s", err)
	}
	log.Info("kafka-cluster: consumer initialized without error")

	producer, err := sarama.NewSyncProducerFromClient(client)
	if err != nil {
		log.Fatalf("kafka-cluster: failed to initialize producer: %s", err)
	}

	c := NotifierKafka{
		instance: instance,
		in:       make(chan mdata.SavedChunk),
		bPool:    util.NewBufferPool(),
		handler:  handler,
		client:   client,
		consumer: consumer,
		producer: producer,

		StopChan:      make(chan int),
		stopConsuming: make(chan struct{}),
	}
	c.start()
	go c.produce()

	return &c
}

func (c *NotifierKafka) start() {
	pre := time.Now()
	processBacklog := new(sync.WaitGroup)
	for _, partition := range partitions {
		var offsetTime int64
		switch offsetStr {
		case "oldest":
			offsetTime = sarama.OffsetOldest
		case "newest":
			offsetTime = sarama.OffsetNewest
		default:
			offsetTime = time.Now().Add(-1*offsetDuration).UnixNano() / int64(time.Millisecond)
		}
		startOffset, err := c.client.GetOffset(topic, partition, offsetTime)
		if err != nil {
			offsetTime = sarama.OffsetOldest
			log.Warnf("kafkamdm: failed to get offset %s: %s -> will use oldest instead", offsetDuration, err)
		}
		processBacklog.Add(1)
		go c.consumePartition(topic, partition, startOffset, processBacklog)
	}
	// wait for our backlog to be processed before returning.  This will block metrictank from consuming metrics until
	// we have processed old metricPersist messages. The end result is that we wont overwrite chunks in cassandra that
	// have already been previously written.
	// We don't wait more than backlogProcessTimeout for the backlog to be processed.
	log.Info("kafka-cluster: waiting for metricPersist backlog to be processed.")
	backlogProcessed := make(chan struct{}, 1)
	go func() {
		processBacklog.Wait()
		backlogProcessed <- struct{}{}
	}()
{'fraction_non_alphanumeric': 0.10715526224383466, 'fraction_numerical': 0.0027787426189649182, 'mean_word_length': 4.114564831261101, 'pattern_counts': {'":': 4, '<': 6, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5698303', 'n_tokens_mistral': 2026, 'n_tokens_neox': 1880, 'n_words': 530}