Heka Kafka to file

hekakafka2file.toml

[hekad]
maxprocs = 8

[Kafka]
type = "KafkaInput"
topic = "mmbb"
addrs = ["kafka0:9092","kafka1:9092","kafka2:9092"]
decoder = "AccesslogDecoder"
offset_method = "Oldest"

[AccesslogDecoder]
type = "PayloadRegexDecoder"
match_regex = '(?P<gollumdate>\d[^\s]+)\s+(?P<local_ip>[^\s]+?)\s+(?P<hostname>[^\s]+?)\s+(?P<private_ip>.*?),?\s*(?P<forwarded>[^\s]+?)\s-\s(?P<Timestamp>\d[^\s]+) \"(?P<method>\w+) (?P<url_path>[^\?]+)(?:\?(?P<query>[^\s]*?)\s+)?HTTP\/(?P<http_version>[^\s]+?)\" (?P<code>\d+) (-|(?P<size>\d+)) (-|(?P<response_time>\d+))\s+(?P<page_id>[^\s]+?)\s+(-|(?P<db>\d+))\s+(-|(?P<java>\d+))\s+(-|(?P<memcache>\d+))\s+(-|(?P<xcache>\d+))\s+(-|(?P<ftp>\d+))\s+(-|(?P<solr>\d+))\s+(-|(?P<redis>\d+))\s+\"-?(?:(http|https):\/\/(?P<referer_host>[^\/]+)\/?\??(?P<referer_query>[^\s]*?))?\" \"(?P<unique_ident>[^\s]+?)\"\s+\"(?P<user_agent>.+)\"'
timestamp_layout= '20060102150405'

[AccesslogDecoder.message_fields]
gollum_date = "%gollumdate%"
local_ip = "%local_ip%"
hostname = "%hostname%"
private_ip = "%private_ip%"
forwarded = "%forwarded%"
timestamp = "%Timestamp%"
method = "%method%"
url_path = "%url_path%"
query = "%query%"
http_version = "%http_version%"
code = "%code%"
size = "%size%"
response_time = "%response_time%"
page_id = "%page_id%"
db = "%db%"
java = "%java%"
memcache = "%memcache%"
xcache = "%xcache%"
ftp = "%ftp%"
solr = "%solr%"
redis = "%redis%"
referer_host = "%referer_host%"
referer_query = "%referer_query%"
unique_ident = "%unique_ident%"
user_agent = "%user_agent%"

[PayloadEncoder]
append_newlines = false

[ESJsonEncoder]
index = "%{Type}-%{2006.01.02}"
es_index_from_timestamp = true
type_name = "%{Type}"

[ElasticSearchOutput]
message_matcher = "TRUE"
server = "http://elastic0:9200"
flush_interval = 5000
flush_count = 10
encoder = "ESJsonEncoder"