in_kafka: Emit each record with its own tag when tag_source is record

[Watson1978]

in_kafka accepts tag_source record but never routes by it. consume collects the whole fetch batch into one MultiEventStream while overwriting tag in the message loop, so the single emit_stream call at the end uses the tag of the last parsed record. Every record of the batch is emitted with that tag, and records that carry a different tag field are silently routed to the wrong destination.

This groups the records by tag and emits one stream per tag, the same way in_kafka_group#process_batch_with_record_tag already works.

fluent-plugin-kafka/lib/fluent/plugin/in_kafka_group.rb

Lines 267 to 310 in e6be8f1

	def process_batch_with_record_tag(batch)
	es = {}
	batch.messages.each { |msg|
	begin
	record = @parser_proc.call(msg)
	tag = record[@record_tag_key]
	tag = @add_prefix + "." + tag if @add_prefix
	tag = tag + "." + @add_suffix if @add_suffix
	es[tag] ||= Fluent::MultiEventStream.new
	case @time_source
	when :kafka
	record_time = Fluent::EventTime.from_time(msg.create_time)
	when :now
	record_time = Fluent::Engine.now
	when :record
	if @time_format
	record_time = @time_parser.parse(record[@record_time_key].to_s)
	else
	record_time = record[@record_time_key]
	end
	else
	log.fatal "BUG: invalid time_source: #{@time_source}"
	end
	if @kafka_message_key
	record[@kafka_message_key] = msg.key
	end
	if @add_headers
	msg.headers.each_pair { |k, v|
	record[k] = v
	}
	end
	es[tag].add(record_time, record)
	rescue => e
	log.warn "parser error in #{batch.topic}/#{batch.partition}", :error => e.to_s, :value => msg.value, :offset => msg.offset
	log.debug_backtrace
	end
	}
	unless es.empty?
	es.each { |tag,es|
	emit_events(tag, es)
	}
	end
	end

Records whose tag field is not a String, or is an empty string, are now skipped with a warning. Emitting per tag makes one such record raise out of emit_stream before the remaining streams are emitted, so the offset stays where it is and the earlier tags are emitted again on every interval. The offset now also advances when every message of a batch is skipped, which used to leave @next_offset unchanged and fetch the same batch on every interval.

The default tag_source topic path is unchanged and still emits a single stream.