lib/fluent/plugin/out_kafka_buffered.rb in fluent-plugin-kafka-enchanced-0.5.24 vs lib/fluent/plugin/out_kafka_buffered.rb in fluent-plugin-kafka-enchanced-0.5.25
- old
+ new
@@ -200,13 +200,14 @@
require 'avro_turf/messaging'
require "avro/builder"
Proc.new do |tag, time, record|
record = record.select{|key, value| !key.nil? && !key.empty?}.map{|k, v| [k.tr('[]-', '_').delete('$'), (v.is_a?(String) ? v.to_s.force_encoding("UTF-8") : v)]}.to_h
timestamp = Time.new
- record['enchilada_timestamp'] = timestamp.strftime('%s%3N').to_i
record['enchilada_time_with_format'] = timestamp.strftime("%Y-%m-%dT%H:%M:%S.%LZ")
fields = record.map{|key, value| {'name' => key, 'type' => (value.is_a?(Fixnum) ? 'int' : (value.is_a?(Float) ? 'float' : 'string'))}}
+ record['enchilada_timestamp'] = timestamp.strftime('%s%3N').to_i
+ fields << {"name" => "enchilada_timestamp", "type" => "long"}
@topic_name = schema_name = "#{tag.to_s.tr('.$:', '_')}_#{Digest::MD5.new.hexdigest(fields.to_s)[0..5]}"
schema_json = {
"type": "record",
"name": schema_name,
"fields": fields