I am trying to consume from an MSK cluster using logstash and push the messages to an Elasticsearch instance. Due to some reason the data is not getting fetched correctly, and getting below message in logs.
Skipping fetch for partition {TOPIC}-0 because previous request to (id: 2 rack: use1-az2) has not been processed.
Below is my logstash config.
input {
kafka {
bootstrap_servers => "{MSK_HOST}"
topics => "{TOPIC}"
group_id => "${KAFKA_GRP_ID}"
auto_offset_reset => "latest"
#ssl_truststore_location => "/usr/share/logstash/config/kafka.client.truststore.jks"
security_protocol => "SSL"
decorate_events => true
codec => json
}
}
filter {
ruby {
code => "event.set('[@metadata][kafka][lc_topic]', event.get('[@metadata][kafka][topic]').split(/(?=[A-Z])/).map{|x| x.downcase }.join('') )"
}
}
output {
amazon_es {
hosts => ["${ES_HOST}"]
ssl => true
#ilm_enabled => false
index => "logstash.%{[@metadata][kafka][lc_topic]}.%{+YYYY.MM.dd}"
manage_template => false
aws_access_key_id => ''
aws_secret_access_key => ''
}
stdout {
codec => rubydebug
}
}