fluent.conf

<source>
  @type kafka_group
  brokers "#{ENV['KAFKA_BROKERS']}"
  consumer_group "#{ENV['KAFKA_CONSUMER_GROUP']}"
  topics "#{ENV['KAFKA_TOPIC']}"
</source>

<filter search>
  @type mysql_json
</filter>

<match search>
  @type mysql_bulk
  host "#{ENV['DATABASE_HOSTNAME']}"
  database "#{ENV['DATABASE_NAME']}"
  username "#{ENV['DATABASE_USERNAME']}"
  password "#{ENV['DATABASE_PASSWORD']}"
  column_names "#{ENV['DATABASE_COLUMNNAMES']}"
  key_names "#{ENV['DATABASE_KEY_NAMES']}"
  table "#{ENV['DATABASE_TABLE']}"
  
    <buffer>
    @type memory
    flush_mode interval
    flush_interval 10s
    retry_max_interval 3
    retry_max_times 5s
    retry_timeout 10s
    overflow_action drop_oldest_chunk
  </buffer>
  
</match>