input {
kafka{
bootstrap_servers => ["kafka-cn-internet.aliyun.com:8080"] #kafka系统的连接地址
client_id => ‘tt‘ #客户端上传到es时，新增字段
group_id => "CID-LOG" #kafka分组的信息
auto_offset_reset => "latest" #从最新的偏移量开始消费
consumer_threads => 5
decorate_events => true #此属性会将当前topic、offset、group、partition等信息也带到message中
topics => ["alikafka-cid-log"] #//数组类型，可配置多个topic 跟
type => "bhy" #//所有插件通用属性,尤其在input里面配置多个数据源时很有用
security_protocol => "SASL_SSL" #kafka连接阿里云的协议
sasl_mechanism => "ONS" #kafka阿里云的消费机制， logstash中默认的是 GSSAPI
jaas_path => "/data/logstash/config/kafka_client_jaas.conf" # ONS登录信息的路径
ssl_keystore_location => ‘/data/logstash/config/kafka.client.truststore.jks‘ #证书
ssl_truststore_location => ‘/data/logstash/config/kafka.client.truststore.jks‘#信任证书
ssl_keystore_password => "xxxx" #证书密码
ssl_truststore_password => "xxxx" #证书密码
codec => json {
charset => "UTF-8"
}
}
}