<commit_before>package worker

import (
	"github.com/Sirupsen/logrus"
	"github.com/confluentinc/confluent-kafka-go/kafka"
	"sync"
	"time"
)

type workerThread struct {
	consumerWorker    *ConsumerWorker
	id                int64
	logger            *logrus.Entry // From Worker
	topic             string        // From Worker
	stopChan          chan bool
	consumer          *kafka.Consumer
	waitGroup         *sync.WaitGroup
	statsMutex        *sync.Mutex
	errors            int64
	succeed           int64
	reportStatsTicker *time.Ticker
}

func newWorkerThread(w *ConsumerWorker, index int64) *workerThread {
	return &workerThread{
		consumerWorker:    w,
		id:                (index + 1),
		stopChan:          make(chan bool),
		waitGroup:         w.waitGroup,
		topic:             w.topic,
		statsMutex:        &sync.Mutex{},
		errors:            0,
		succeed:           0,
		reportStatsTicker: time.NewTicker(30 * time.Second),
	}
}

func (wt *workerThread) stop() {
	wt.reportStats()
	close(wt.stopChan)
}

func (wt *workerThread) start() {

	c, err := kafka.NewConsumer(&kafka.ConfigMap{
		"bootstrap.servers":               wt.consumerWorker.broker,
		"group.id":                        wt.consumerWorker.group,
		"session.timeout.ms":              6000,
		"go.events.channel.enable":        true,
		"go.application.rebalance.enable": true,
		"default.topic.config":            kafka.ConfigMap{"auto.offset.reset": "earliest"}})

	if err != nil {<commit_msg>change worker thread report interval from 30 seconds to 1 minute<commit_after>package worker

import (
	"github.com/Sirupsen/logrus"
	"github.com/confluentinc/confluent-kafka-go/kafka"
	"sync"
	"time"
)

type workerThread struct {
	consumerWorker    *ConsumerWorker
	id                int64
	logger            *logrus.Entry // From Worker
	topic             string        // From Worker
	stopChan          chan bool
	consumer          *kafka.Consumer
	waitGroup         *sync.WaitGroup
	statsMutex        *sync.Mutex
	errors            int64
	succeed           int64
	reportStatsTicker *time.Ticker
}

func newWorkerThread(w *ConsumerWorker, index int64) *workerThread {
	return &workerThread{
		consumerWorker:    w,
		id:                (index + 1),
		stopChan:          make(chan bool),
		waitGroup:         w.waitGroup,
		topic:             w.topic,
		statsMutex:        &sync.Mutex{},
		errors:            0,
		succeed:           0,
		reportStatsTicker: time.NewTicker(60 * time.Second),
	}
}

func (wt *workerThread) stop() {
	wt.reportStats()
	close(wt.stopChan)
}

func (wt *workerThread) start() {

	c, err := kafka.NewConsumer(&kafka.ConfigMap{
		"bootstrap.servers":               wt.consumerWorker.broker,
		"group.id":                        wt.consumerWorker.group,
		"session.timeout.ms":              6000,
		"go.events.channel.enable":        true,
		"go.application.rebalance.enable": true,
		"default.topic.config":            kafka.ConfigMap{"auto.offset.reset": "earliest"}})

	if err != nil {
{'fraction_non_alphanumeric': 0.11017803157541149, 'fraction_numerical': 0.012428619415518979, 'mean_word_length': 2.1987110633727176, 'pattern_counts': {'":': 14, '<': 3, '<?xml version=': 0, '>': 3, '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': '5075253', 'n_tokens_mistral': 1018, 'n_tokens_neox': 940, 'n_words': 200}