<commit_before>﻿/*
 * Created by SharpDevelop.
 * User: peng.zang
 * Date: 11/11/2014
 * Time: 10:49 AM
 * 
 * To change this template use Tools | Options | Coding | Edit Standard Headers.
 */
using System;
using System.Linq;
using System.Collections.Generic;

using KafkaNet.Protocol;
using KafkaNet.Model;

namespace KafkaNet
{
	/// <summary>
	/// Description of NativeHLConsumer. Should only be used per
	/// </summary>
	public class NativeHLConsumer : Consumer
	{

		protected string _consumerGroup;
		
		public NativeHLConsumer(ConsumerOptions options, string consumerGroup, params OffsetPosition[] positions) : base(options, positions){
			if(_topic == null || _topic.Name != _options.Topic)
				_topic = _metadataQueries.GetTopic(_options.Topic);
			_consumerGroup = consumerGroup;
			RefreshOffsets();
		}
		
		public void RefreshOffsets(){
			_topic.Partitions.ForEach(
				partition => {
					var conn = _options.Router.SelectBrokerRoute(_topic.Name, partition.PartitionId);
					conn.Connection
						.SendAsync(CreateOffsetFetchRequest(_consumerGroup, partition.PartitionId))
						.Result.ForEach(
							offsetResp => {
								Console.WriteLine("fetch offset: " + offsetResp.ToString());
								_partitionOffsetIndex.AddOrUpdate(partition.PartitionId, i => offsetResp.Offset, (i, l) => offsetResp.Offset);
							});
				});
			
//			var actualOffsets = _metadataQueries.GetTopicOffsetAsync(_options.Topic).Result;
//			var maxGroup = from o in actualOffsets 
//						group o by o.PartitionId into g 
//				select new {pid = g.Key, offset = g.Max(x => x.)};

		}
		
		public IEnumerable<Message> Consume(int num){
//			RefreshOffsets();
			var result = base.Consume(null).Take(num).ToList();
//			var maxgroups = result.GroupBy(x => x.Meta.PartitionId).GroupJoin(;
			var maxgroups = from r in result
				group r by r.Meta.PartitionId into g 
				select new {pid = g.Key, offset = g.Max(m => m.Meta.Offset) + 1 };
			
			maxgroups.ToList().ForEach(x => Console.WriteLine(x.pid + " : " + x.offset));
			
			foreach (var pos in maxgroups) {
//				Console.WriteLine("partition: " + pos.Meta.PartitionId + " , offset: "+ pos.Meta.Offset);
				
				if(!CommitOffset(pos.pid, pos.offset)){
					var mingroup = from r in result 
						group r by r.Meta.PartitionId into g 
						select new { pid = g.Key, offset = g.Min(m => m.Meta.Offset)};
					mingroup.ToList().ForEach(x => CommitOffset(x.pid, x.offset));
					return null;
				}
			}
			return result;
		}
		
		protected bool CommitOffset(int pid, long offset){
			Console.WriteLine("*** Committing partition: " + pid + ", offset: " + offset);
			var conn = _options.Router.SelectBrokerRoute(_topic.Name, pid);
			var resp = conn.Connection
				.SendAsync(CreateOffsetCommitRequest(_consumerGroup, pid, offset)).Result.FirstOrDefault();<commit_msg>check offset for consumer group if it's in a correct range; if not, correct it by setting it to minimum offset.<commit_after>﻿/*
 * Created by SharpDevelop.
 * User: peng.zang
 * Date: 11/11/2014
 * Time: 10:49 AM
 * 
 * To change this template use Tools | Options | Coding | Edit Standard Headers.
 */
using System;
using System.Linq;
using System.Collections.Generic;

using KafkaNet.Protocol;
using KafkaNet.Model;

namespace KafkaNet
{
	/// <summary>
	/// Description of NativeHLConsumer. Should only be used per
	/// </summary>
	public class NativeHLConsumer : Consumer
	{

		protected string _consumerGroup;
		
		public NativeHLConsumer(ConsumerOptions options, string consumerGroup, params OffsetPosition[] positions) : base(options, positions){
			if(_topic == null || _topic.Name != _options.Topic)
				_topic = _metadataQueries.GetTopic(_options.Topic);
			_consumerGroup = consumerGroup;
			RefreshOffsets();
		}
		
		public void RefreshOffsets(){
			var actualOffsets = _metadataQueries.GetTopicOffsetAsync(_options.Topic).Result;
			var maxminGroups = actualOffsets.Select(x => new {pid = x.PartitionId, min = x.Offsets.Min(), max = x.Offsets.Max()});
			
			_topic.Partitions.ForEach(
				partition => {
					var conn = _options.Router.SelectBrokerRoute(_topic.Name, partition.PartitionId);
					conn.Connection
						.SendAsync(CreateOffsetFetchRequest(_consumerGroup, partition.PartitionId))
						.Result.ForEach(
							offsetResp => {
								Console.WriteLine("fetch offset: " + offsetResp.ToString());
								
								if(actualOffsets.Any(x => x.PartitionId==partition.PartitionId)){
									var actual = maxminGroups.First(x => x.pid==partition.PartitionId);
									if(actual.min > offsetResp.Offset || actual.max < offsetResp.Offset){
										offsetResp.Offset = actual.min;
									}
								}
								_partitionOffsetIndex.AddOrUpdate(partition.PartitionId, i => offsetResp.Offset, (i, l) => offsetResp.Offset);
							});
				});
			
			
//				group o by o.PartitionId into g
//				select new {pid = g.Key, min = g.Min(x => x.Offsets), max = g.Max(x => x.Offsets) };
			

		}
		
		public IEnumerable<Message> Consume(int num){
//			RefreshOffsets();
			var result = base.Consume(null).Take(num).ToList();
//			var maxgroups = result.GroupBy(x => x.Meta.PartitionId).GroupJoin(;
			var maxgroups = from r in result
				group r by r.Meta.PartitionId into g
				select new {pid = g.Key, offset = g.Max(m => m.Meta.Offset) + 1 };
			
			maxgroups.ToList().ForEach(x => Console.WriteLine(x.pid + " : " + x.offset));
			
			foreach (var pos in maxgroups) {
//				Console.WriteLine("partition: " + pos.Meta.PartitionId + " , offset: "+ pos.Meta.Offset);
				
				if(!CommitOffset(pos.pid, pos.offset)){
					var mingroup = from r in result
						group r by r.Meta.PartitionId into g
						select new { pid = g.Key, offset = g.Min(m => m.Meta.Offset)};
					mingroup.ToList().ForEach(x => CommitOffset(x.pid, x.offset));
					return null;
				}
			}
			return result;
		}
		
		protected bool CommitOffset(int pid, long offset){
			Console.WriteLine("*** Committing partition: " + pid + ", offset: " + offset);
			var conn = _options.Router.SelectBrokerRoute(_topic.Name, pid);
			var resp = conn.Connection
				.SendAsync(CreateOffsetCommitRequest(_consumerGroup, pid, offset)).Result.FirstOrDefault();
{'fraction_non_alphanumeric': 0.1306401694086985, 'fraction_numerical': 0.004235217462127383, 'mean_word_length': 3.3361581920903953, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 34, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7038346', 'n_tokens_mistral': 2293, 'n_tokens_neox': 1864, 'n_words': 456}