<commit_before>﻿using System;
using System.Linq;
using System.Threading;
using Kafka.Basic;
using Metrics;

namespace Consumer
{
    class BatchedConsumer
    {
        private Thread _consoleThread;

        public int Start(BatchedConsumerOptions opts)
        {
            var timer = Metric.Timer("Received", Unit.Events);
            Metric.Config.WithReporting(r => r.WithConsoleReport(TimeSpan.FromSeconds(5)));

            using (var client = new KafkaClient(opts.ZkConnect))
            using (var consumer = new Kafka.Basic.BatchedConsumer(client, opts.Group, opts.Topic, opts.BatchSizeMax, opts.BatchTimeoutMs))
            {
                ListenToConsole(consumer);
                Thread.Sleep(5000);
                consumer
                    .Start(m =>
                    {
                        var time = DateTime.UtcNow.Ticks;
                        m.ToList()
                            .ForEach(message =>
                            {
                                var value = long.Parse(message.Value);
                                var diff = (time - value) / 10000;
                                timer.Record(diff, TimeUnit.Milliseconds);
                            });
                    }
                    );
            }

            return 0;
        }

        private void ListenToConsole(Kafka.Basic.BatchedConsumer consumer)
        {
            _consoleThread = new Thread(() =>
            {
                Console.CancelKeyPress += (sender, eventArgs) =>
                {
                    Console.WriteLine("Kill!");
                    _consoleThread.Abort();
                };
                while (true)
                {
                    var input = Console.ReadKey(true);
                    //if (input.KeyChar == 'p')
                    //{
                    //    stream.Pause();
                    //    Console.WriteLine("Paused.");
                    //}
                    //if (input.KeyChar == 'r')
                    //{
                    //    stream.Resume();
                    //    Console.WriteLine("Resumed.");
                    //}
                    if (input.KeyChar == 'q')
                    {
                        Console.WriteLine("Shutting down...");
                        consumer.Shutdown();
                        break;
                    }
                }
            });
            _consoleThread.Start();
        }
    }
}
<commit_msg>use a histogram to report batch size metrics<commit_after>﻿using System;
using System.Linq;
using System.Threading;
using Kafka.Basic;
using Metrics;

namespace Consumer
{
    class BatchedConsumer
    {
        private Thread _consoleThread;

        public int Start(BatchedConsumerOptions opts)
        {
            var histogram = Metric.Histogram("batch.size", Unit.Items);
            var timer = Metric.Timer("message.latency", Unit.Events);
            Metric.Config.WithReporting(r => r.WithConsoleReport(TimeSpan.FromSeconds(5)));

            using (var client = new KafkaClient(opts.ZkConnect))
            using (var consumer = new Kafka.Basic.BatchedConsumer(client, opts.Group, opts.Topic, opts.BatchSizeMax, opts.BatchTimeoutMs))
            {
                ListenToConsole(consumer);

                consumer
                    .Start(m =>
                    {
                        var time = DateTime.UtcNow.Ticks;
                        var list = m.ToList();
                        list
                            .ForEach(message =>
                            {
                                var value = long.Parse(message.Value);
                                var diff = (time - value) / 10000;
                                timer.Record(diff, TimeUnit.Milliseconds);
                            });
                        histogram.Update(list.Count);
                    });
            }

            return 0;
        }

        private void ListenToConsole(Kafka.Basic.BatchedConsumer consumer)
        {
            _consoleThread = new Thread(() =>
            {
                Console.CancelKeyPress += (sender, eventArgs) =>
                {
                    Console.WriteLine("Kill!");
                    _consoleThread.Abort();
                };
                while (true)
                {
                    var input = Console.ReadKey(true);
                    //if (input.KeyChar == 'p')
                    //{
                    //    stream.Pause();
                    //    Console.WriteLine("Paused.");
                    //}
                    //if (input.KeyChar == 'r')
                    //{
                    //    stream.Resume();
                    //    Console.WriteLine("Resumed.");
                    //}
                    if (input.KeyChar == 'q')
                    {
                        Console.WriteLine("Shutting down...");
                        consumer.Shutdown();
                        break;
                    }
                }
            });
            _consoleThread.Start();
        }
    }
}

{'fraction_non_alphanumeric': 0.10076756543987404, 'fraction_numerical': 0.0035426097224955716, 'mean_word_length': 0.9546153846153846, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5613354', 'n_tokens_mistral': 1231, 'n_tokens_neox': 1121, 'n_words': 208}