from kafka import KafkaConsumer
import threading

TOPIC_NAME = 'お客様の消費トピック'
GROUP_ID = 'お客様のコンシューマーグループ名'
# サービスアドレス+ポート。外部ネットワークポートは9096、内部ネットワークポートは9095です。例は内部ネットワークでの消費です。お客様の実際の状況に応じて入力してください。
BOOTSTRAP_SERVERS = ''kafkaconsumer-${region}.cls.tencentyun.com:9095''

def consume_messages(thread_id):
    # Kafkaコンシューマーインスタンスを作成します。
    consumer = KafkaConsumer(
        TOPIC_NAME,
        group_id=GROUP_ID,
        bootstrap_servers=BOOTSTRAP_SERVERS,
        value_deserializer=lambda m: m.decode('utf-8'),
        auto_offset_reset='earliest',
        security_protocol = "SASL_PLAINTEXT",
        sasl_mechanism = 'PLAIN',
        sasl_plain_username = "${logsetID}",
        sasl_plain_password = "${SecretId}#${SecretKey}",
        api_version = (2, 5, 1)
    )

    try:
        for message in consumer:
            print(f"Thread {thread_id}: partition = {message.partition}, offset = {message.offset}, value = {message.value}")
    except KeyboardInterrupt:
        pass
    finally:
        # コンシューマーを閉じます。
        consumer.close()

if __name__ == "__main__":
    # 3つのコンシューマースレッドを起動します。これは例です。お客様の実際の状況に応じて設定してください。
    num_consumers = 3 
    threads = []
    for i in range(num_consumers):
        thread = threading.Thread(target=consume_messages, args=(i,))
        threads.append(thread)
        thread.start()

    # すべてのスレッドが終了するのを待ちます。
    for thread in threads:
        thread.join()

<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.5.0</version>
</dependency>
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerGroupTest {

    public static void consume() {
        Properties props = new Properties();
        String logset_id = "${logsetID}";
        // CLSコンソールのkafkaプロトコル消費ページに表示されるトピック名
        String topic_id = "お客様の消費トピック";

        String accessKeyID = System.getenv("${SecretId}");
        String accessKeySecret = System.getenv("${SecretKey}");

        String groupId = "お客様のコンシューマーグループ名";

       // サービスアドレス+ポート。外部ネットワークポートは9096、内部ネットワークポートは9095です。例は内部ネットワークでの消費です。お客様の実際の状況に応じて入力してください。
        String hosts = "kafkaconsumer-${region}.cls.tencentyun.com:9095";
        props.put("bootstrap.servers", hosts);
        props.put("security.protocol", "SASL_PLAINTEXT");
        props.put("sasl.mechanism", "PLAIN");
        props.put("sasl.jaas.config",
       "org.apache.kafka.common.security.plain.PlainLoginModule required username=\\"" +
        logset_id + "\\" password=\\"" + accessKeyID + "#" + accessKeySecret + "\\";");

        // Kafkaコンシューマー設定
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "5000");
        props.put("session.timeout.ms", "10000");
        props.put("auto.offset.reset", "earliest");
        props.put("max.poll.interval.ms", "120000");
        props.put("heartbeat.interval.ms", "3000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // Kafkaコンシューマーインスタンスを作成します。
        KafkaConsumer<String,String> consumer =  new KafkaConsumer<String,String>(props);
        consumer.subscribe(Collections.singletonList(topic_id));
        while(true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Received message: (" + record.key() + ", " + record.value() + ") at offset " + record.offset());
            }
        }
    }
    public static void main(String[] args){
        consume();
    }
}