filebeat.inputs:
- type: kafka
  hosts:
    - kafkaconsumer-${region}.cls.tencentyun.com:9095
  topics: "您的消费主题"
  group_id: "您的消费组名称"
  username: "${logsetID}"
  password: "${SecretId}#${SecretKey}"
  sasl.mechanism: "PLAIN" 
  processors:
    - decode_json_fields:
        fields: ["message"]
        target: ""
        overwrite_keys: true
output.file:
  path: /tmp
  filename: filebeat_data.log
  rotate_every_kb: 102400
  number_of_files: 7


input {
    kafka {
        # cls kafka 协议消费控制台给出的的主题名称，例如 XXXXXX-633a268c-XXXX-4a4c-XXXX-7a9a1a7baXXXX，可在控制台复制
        topics => "您的消费主题"
        # 服务地址+端口，外网端口9096，内网端口9095，例子是内网消费，请根据您的实际情况填写
        bootstrap_servers => "kafkaconsumer-${region}.cls.tencentyun.com:9095"
        group_id => "您的消费组名称"
        security_protocol => "SASL_PLAINTEXT"
        sasl_mechanism => "PLAIN"
        # 用户名是日志集合 ID，例如 ca5cXXXXdd2e-4ac0af12-92d4b677d2c6 
        # 密码是用户的 SecretId#SecretKey 组合的字符串，例如 AKID********************************#XXXXuXtymIXT0Lac，注意不要丢失#。建议使用子账号密钥，主账号为子账号授权时，遵循最小权限原则，即子账号的访问策略中的 action、resource 都配置为最小范围，可以满足操作即可。
        sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='${logsetID}' password='${SecretId}#${SecretKey}';"
    }
    
}
output {
  stdout { codec => json }
}


a1.sources = source_kafka
a1.sinks = sink_local
a1.channels = channel1

# 配置Source
a1.sources.source_kafka.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.source_kafka.batchSize = 10
a1.sources.source_kafka.batchDurationMillis = 200000
# 服务地址+端口，外网端口9096，内网端口9095，例子是内网消费，请根据您的实际情况填写
a1.sources.source_kafka.kafka.bootstrap.servers = kafkaconsumer-${region}.cls.tencentyun.com:9095
# cls kafka 协议消费控制台给出的的主题名称，例如 XXXXXX-633a268c-XXXX-4a4c-XXXX-7a9a1a7baXXXX，可在控制台复制
a1.sources.source_kafka.kafka.topics = 您的消费主题  
# 请替换为您的消费组名称
a1.sources.source_kafka.kafka.consumer.group.id = 您的消费组名称
a1.sources.source_kafka.kafka.consumer.auto.offset.reset = earliest
a1.sources.source_kafka.kafka.consumer.security.protocol = SASL_PLAINTEXT
a1.sources.source_kafka.kafka.consumer.sasl.mechanism = PLAIN
# 用户名是日志集合 ID，例如 ca5cXXXXdd2e-4ac0af12-92d4b677d2c6
# 密码是用户的 SecretId#SecretKey 组合的字符串，例如 AKID********************************#XXXXuXtymIXT0Lac注意不要丢失#。建议使用子账号密钥，主账号为子账号授权时，遵循最小权限原则，即子账号的访问策略中的 action、resource 都配置为最小范围，可以满足操作即可，注意 jaas.config 最后有;分号，不填写会报错。
a1.sources.source_kafka.kafka.consumer.sasl.jaas.config = org.apache.kafka.common.security.plain.PlainLoginModule required username="${logsetID}" password="${SecretId}#${SecretKey}";

# 配置 sink
a1.sinks.sink_local.type = logger

a1.channels.channel1.type = memory
a1.channels.channel1.capacity = 1000
a1.channels.channel1.transactionCapacity = 100

# 将 source 和 sink 绑定到 channel
a1.sources.source_kafka.channels = channel1
a1.sinks.sink_local.channel = channel1