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