1_kafka_input_template.conf

# Copyright [2021] [Cargill, Incorporated.] 
# SPDX-License-Identifier: Apache-2.0
input {
  kafka {
    bootstrap_servers => "VAR_KAFKA_BOOTSTRAP_SERVERS"
    jaas_path => "VAR_KAFKA_JAAS_PATH"
    client_id => "VAR_KAFKA_CLIENT_ID"
    group_id => "VAR_KAFKA_GROUP_ID"
    ssl_truststore_location => "VAR_KAFKA_CLIENT_TRUSTSTORE"
    ssl_truststore_password => "VAR_KAFKA_TRUSTSTORE_PASSWORD"
    client_rack => "VAR_RACK_ID"
    security_protocol => "SASL_SSL"
    sasl_mechanism => "SCRAM-SHA-512"
    topics => ["VAR_KAFKA_TOPIC"]
    id => "VAR_LOGSTASH_PLUGIN_ID"
    max_poll_records => VAR_MAX_POLL_RECORDS
    consumer_threads => VAR_CONSUMER_THREADS
    partition_assignment_strategy => "cooperative_sticky"
    auto_commit_interval_ms => "1000"
    auto_offset_reset => "latest"
    connections_max_idle_ms => "540000"
    receive_buffer_bytes => "2097152"
    reconnect_backoff_ms => "200"
    max_poll_interval_ms => "300000"
    request_timeout_ms => "40000"
    poll_timeout_ms => "2000"
    heartbeat_interval_ms => "1000"
    session_timeout_ms => "10000"
    retry_backoff_ms => "1000"
    codec => "VAR_CODEC"
  }
}
filter {
  mutate {
    # VAR_CUSTOM_FIELDS
  }
  ruby {
    code => "
      if !event.get('host').is_a?(Hash)
        event.set('[agent][name]', event.get('host'))
      end
    "
  }
}
output {
  pipeline { send_to => [VAR_PIPELINE_NAME]}
}