1) 3.6.0 ver 설치가이드(240111)

운영환경 : centOS 7

• 설치 전 환경 확인 (root 유저로 수행)
• 설치 기본 환경

OSJAVAKAFKAZOOKEEPER

Centos 7 *파일디스크립터는 100,000 이상 권장

-Kafka 2.0.0 이상의 버전은 java 7이하 버전 지원 불가 -Kafka 2.1.0 이상의 버전부터 java11 버전도 지원

-6GB 이하의 heap메모리 권장 -브로커당 2000개를 max partition으로 권장 -heap 메모리로 설정하고 남아있는 메모리는 페이지 캐시로 사용하기를 권장

-호스트의 메모리는 최소 8GB 권장 -heap 사이즈는 1GB 권장

 

---

 

[ JDK 설치 ]

$> yum install java-1.8.0-openjdk $>java –version

 

[ kafka 설치 ]

• 유저 생성 (root 유저로 수행)

      -  kafka 유저 생성

$> useradd kafka $> passwd kafka

• kafka directory 생성 및 권한 변경 ( root 권한으로 진행 ) 

$> cd / $> mkdir kafka $> chown –R kafka:kafka kafka

• JAVA_HOME 지정 ( kafka 권한으로 진행 )

$> su kafka # kafka 유저로 전환 $> cd /home/kafka $> vi .bash_profile # EXPORT JAVA_HOME export JAVA_HOME=”/usr/lib/jvm/java-11” #jdk8버전인 경우 java-1.8.0로 수정 export PATH=$JAVA_HOME/bin:$PATH

 

• 설치 파일 다운로드 ( kafka 유저로 진행 )

       - kafk, zookeeper 설치 파일 다운로드

$> cd /kafka $> wget https://downloads.apache.org/kafka/3.6.0/kafka_2.13-3.6.0.tgz 환경에 따라 가능한 방법으로 필요한 파일 다운로드 권장

• kafka 설치 파일 압축 해제 ( kafka 유저로 진행 )

$> cd /kafka/  $> tar xvzf kafka_2.13-3.6.0.tgz

• kafka 설정 변경 ( 내부 서버 이용할 경우 )

$> cd /kafka/kafka_2.13-3.6.0/config $> cp -rp server.properties server.properties_YYYYMMDD_orignal  # 백업파일 저장 # 34 line 주석 해제 Listeners=PLANINTEXT://:9092 #broker가 사용하는 호스트와 port를 지정, 기본값은listeners를 사용.  # 38 line 주석 해제 및 변경 advertised.listeners=PLAINTEXT://localhost:9092  #producer와 consumer가 접근할 호스트와 port를 지정. 내부 서버 접근할 경우 localhost 사용

 

• kafka 설정 변경 ( 외부 서버 이용할 경우 )

       - 포트 포워딩 규칙 port 번호 9092 추가

     1. listener에 ip 지정할 경우

$> cd /kafka/kafka_2.13-3.6.0/config $> cp -rp server.properties server.properties_YYYYMMDD_orignal  # 백업파일 저장 # 34 line 주석 해제 Listeners=PLANINTEXT://192.168.56.1:9092  #broker가 바인딩 될 사용자 서버로 변경

    2. advertised.listeners에 ip 지정할 경우

$> cd /kafka/kafka_2.13-3.6.0/config $> cp -rp server.properties server.properties_YYYYMMDD_orignal  # 백업파일 저장 # 34 line 주석 해제 Listeners=PLANINTEXT://:9092 #broker가 사용하는 호스트와 port를 지정 # 38 line 주석 해제 및 변경 advertised.listeners=PLAINTEXT://192.168.56.1:9092  #producer와 consumer가 접근할 호스트와 port를 지정. 외부 서버 접근 할 경우, localhost 대신, 사용자 서버에 맞게 변경 (bootstrap-servers중 반드시 1개와 동일해야 함)

[ zookeeper 프로세스 기동 ]  *반드시 zookeeper 먼저 기동 후, broker 기동

$>cd /kafka/kafka_2.13-3.6.0/bin $> ./zookeeper-server-start.sh -daemon /kafka/kafka_2.13-3.6.0/config/zookeeper.properties # 백그라운드에서 실행 $>ps -ef | grep zookeeper.properties kafka    31225     1  0 11:14 ?        00:00:38 java -Xmx512M -Xms512M -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -XX:MaxInlineLevel=15 -Djava.awt.headless=true -Xloggc:/kafka/kafka_2.13-3.6.0/bin/../logs/zookeeper-gc.log -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=100M - ··· (이하 생략)

[ broker 프로세스 기동 ]

$>./kafka-server-start.sh -daemon /kafka/kafka_2.13-3.6.0/config/server.properties # 백그라운드에서 실행 $>ps -ef | grep server.properties kafka    31225     1  0 11:14 ?        00:00:38 java -Xmx512M -Xms512M -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -XX:MaxInlineLevel=15 -Djava.awt.headless=true -Xloggc:/kafka/kafka_2.13-3.6.0/bin/../logs/zookeeper-gc.log -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=100M - 3.6.0/bin/../logs -Dlog4j.configuration=file:./../config/log4j.properties ··· (이하 생략)

[ kafka 외부 서버 연동 성공 예시 ]

.   ____          _            __ _ _  /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \ ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \  \\/  ___)| |_)| | | | | || (_| |  ) ) ) )   '  |____| .__|_| |_|_| |_\__, | / / / /  =========|_|==============|___/=/_/_/_/  :: Spring Boot ::        (v1.5.6.RELEASE) 2023-12-20 16:29:30.562  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Starting SimpleKafkaProducerApplication on DESKTOP-4553D65 with PID 30856 (C:\Users\KTDS\Downloads\SimpleKafkaExampleSpringBoot-master\SimpleKafkaExampleSpringBoot-master\target\classes started by KTDS in C:\Users\KTDS\Downloads\SimpleKafkaExampleSpringBoot-master) 2023-12-20 16:29:30.566  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : No active profile set, falling back to default profiles: default 2023-12-20 16:29:30.620  INFO 30856 --- [           main] s.c.a.AnnotationConfigApplicationContext : Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@662706a7: startup date [Wed Dec 20 16:29:30 KST 2023]; root of context hierarchy WARNING: An illegal reflective access operation has occurred WARNING: Illegal reflective access by org.springframework.cglib.core.ReflectUtils$1 (file:/C:/Users/KTDS/.m2/repository/org/springframework/spring-core/4.3.10.RELEASE/spring-core-4.3.10.RELEASE.jar) to method java.lang.ClassLoader.defineClass(java.lang.String,byte[],int,int,java.security.ProtectionDomain) WARNING: Please consider reporting this to the maintainers of org.springframework.cglib.core.ReflectUtils$1 WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations WARNING: All illegal access operations will be denied in a future release 2023-12-20 16:29:31.268  INFO 30856 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Registering beans for JMX exposure on startup 2023-12-20 16:29:31.295  INFO 30856 --- [           main] o.a.k.clients.producer.ProducerConfig    : ProducerConfig values: acks = all batch.size = 16384 bootstrap.servers = [192.168.56.105:9092] buffer.memory = 33554432 client.id = compression.type = none connections.max.idle.ms = 540000 enable.idempotence = false interceptor.classes = null key.serializer = class org.apache.kafka.common.serialization.StringSerializer linger.ms = 1 max.block.ms = 60000 max.in.flight.requests.per.connection = 5 max.request..size = 1048576 metadata.max.age.ms = 300000 metric.reporters = [] metrics.num.samples = 2 metrics.recording.level = INFO metrics.sample.window.ms = 30000 partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner receive.buffer.bytes = 32768 reconnect.backoff.max.ms = 1000 reconnect.backoff.ms = 50 request.timeout.ms = 30000 retries = 0 retry.backoff.ms = 100 sasl.jaas.config = null sasl.kerberos.kinit.cmd = /usr/bin/kinit sasl.kerberos.min.time.before.relogin = 60000 sasl.kerberos.service.name = null sasl.kerberos.ticket.renew.jitter = 0.05 sasl.kerberos.ticket.renew.window.factor = 0..8 sasl.mechanism = GSSAPI security.protocol = PLAINTEXT send.buffer.bytes = 131072 ssl.cipher.suites = null ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1] ssl.endpoint.identification.algorithm = null ssl.key.password = null ssl.keymanager.algorithm = SunX509 ssl.keystore.location = null ssl.keystore.password = null ssl.keystore.type = JKS ssl.protocol = TLS ssl.provider = null ssl.secure.random.implementation = null ssl.trustmanager.algorithm = PKIX ssl.truststore.location = null ssl.truststore.password = null ssl.truststore.type = JKS transaction.timeout.ms = 60000 transactional.id = null value.serializer = class org.apache.kafka.common.serialization.StringSerializer 2023-12-20 16:29:31.342  INFO 30856 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 0.11.0.0 2023-12-20 16:29:31.342  INFO 30856 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : cb8625948210849f // AP에서 보내는 메시지 ↓ 2023-12-20 16:29:31.348  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: The index is now: 0 2023-12-20 16:29:31.431  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: The index is now: 1 2023-12-20 16:29:31.431  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: The index is now: 2 2023-12-20 16:29:31.431  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: The index is now: 3 2023-12-20 16:29:31.432  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: The index is now: 4 2023-12-20 16:29:31.432  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: The index is now: 5 2023-12-20 16:29:31.432  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: The index is now: 6 2023-12-20 16:29:31.432  INFO 30856 --- [           main] c.c.p.S..SimpleKafkaProducerApplication   : Sending Kafka message: The index is now: 7 2023-12-20 16:29:31.432  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: The index is now: 8 2023-12-20 16:29:31.432  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: The index is now: 9 2023-12-20 16:29:31.436  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: {"index":0,"message":"The index is now: 0","nestedJsonObject":{"nestedObjectMessage":"This is a nested JSON object with index: 0"}} 2023-12-20 16:29:31.436  INFO 30856 --- [           main] c..c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: {"index":1,"message":"The index is now: 1","nestedJsonObject":{"nestedObjectMessage":"This is a nested JSON object with index: 1"}} 2023-12-20 16:29:31.437  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: {"index":2,"message":"The index is now: 2","nestedJsonObject":{"nestedObjectMessage":"This is a nested JSON object with index: 2"}} 2023-12-20 16:29:31.443  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: {"index":3,"message":"The index is now: 3","nestedJsonObject":{"nestedObjectMessage":"This is a nested JSON object with index: 3"}} 2023-12-20 16:29:31.443  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: {"index":4,"message":"The index is now: 4","nestedJsonObject":{"nestedObjectMessage":"This is a nested JSON object with index: 4"}} 2023-12-20 16:29:31.443  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: {"index":5,"message":"The index is now: 5","nestedJsonObject":{"nestedObjectMessage":"This is a nested JSON object with index: 5"}} 2023-12-20 16:29:31.444  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: {"index":6,"message":"The index is now: 6","nestedJsonObject":{"nestedObjectMessage":"This is a nested JSON object with index: 6"}} 2023-12-20 16:29:31.444  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: {"index":7,"message":"The index is now: 7","nestedJsonObject":{"nestedObjectMessage":"This is a nested JSON object with index: 7"}} 2023-12-20 16:29:31.444  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: {"index":8,"message":"The index is now: 8","nestedJsonObject":{"nestedObjectMessage":"This is a nested JSON object with index: 8"}} 2023-12-20 16:29:31.444  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Sending Kafka message: {"index":9,"message":"The index is now: 9","nestedJsonObject":{"nestedObjectMessage":"This is a nested JSON object with index: 9"}} 2023-12-20 16:29:31.445  INFO 30856 --- [       Thread-2] c.c.p.S.SimpleKafkaProducerApplication   : Starting Kafka consumer thread. 2023-12-20 16:29:31.447  INFO 30856 --- [           main] c.c.p.S.SimpleKafkaProducerApplication   : Started SimpleKafkaProducerApplication in 1.184 seconds (JVM running for 2.309) 2023-12-20 16:29:31.455  INFO 30856 --- [       Thread-2] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values: auto.commit.interval.ms = 1000 auto.offset.reset = latest bootstrap.servers = [192.168.56.105:9092] check.crcs = true client.id = connections.max.idle.ms = 540000 enable.auto.commit = true exclude.internal.topics = true fetch.max.bytes = 52428800 fetch.max.wait.ms = 500 fetch.min..bytes = 1 group.id = thetechcheck heartbeat.interval.ms = 3000 interceptor.classes = null internal.leave.group.on.close = true isolation.level = read_uncommitted key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer max.partition.fetch.bytes = 1048576 max.poll.interval.ms = 300000 max.poll.records = 1 metadata.max.age.ms = 300000 metric.reporters = [] metrics.num.samples = 2 metrics.recording.level = INFO metrics.sample.window.ms = 30000 partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor] receive.buffer.bytes = 65536 reconnect.backoff.max.ms = 1000 reconnect.backoff.ms = 50 request.timeout.ms = 305000 retry.backoff.ms = 100 sasl.jaas.config = null sasl.kerberos.kinit.cmd = /usr/bin/kinit sasl.kerberos.min.time.before.relogin = 60000 sasl.kerberos.service.name = null sasl.kerberos.ticket.renew.jitter = 0.05 sasl.kerberos.ticket..renew.window.factor = 0.8 sasl.mechanism = GSSAPI security.protocol = PLAINTEXT send.buffer.bytes = 131072 session.timeout.ms = 10000 ssl.cipher.suites = null ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1] ssl.endpoint.identification.algorithm = null ssl.key.password = null ssl.keymanager.algorithm = SunX509 ssl.keystore.location = null ssl.keystore.password = null ssl.keystore.type = JKS ssl.protocol = TLS ssl.provider = null ssl.secure.random.implementation = null ssl.trustmanager.algorithm = PKIX ssl.truststore.location = null ssl.truststore.password = null ssl.truststore.type = JKS value.deserializer = class org.apache.kafka.common.serialization..StringDeserializer 2023-12-20 16:29:31.477  WARN 30856 --- [       Thread-2] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'zookeeper.session.timeout.ms' was supplied but isn't a known config. 2023-12-20 16:29:31.478  WARN 30856 --- [       Thread-2] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'zookeeper.sync.time.ms' was supplied but isn't a known config. 2023-12-20 16:29:31.479  WARN 30856 --- [       Thread-2] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'consumer.timeout.ms' was supplied but isn't a known config. 2023-12-20 16:29:31.479  WARN 30856 --- [       Thread-2] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'auto.commit.enable' was supplied but isn't a known config. 2023-12-20 16:29:31.479  INFO 30856 --- [       Thread-2] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 0.11.0.0 2023-12-20 16:29:31.479  INFO 30856 --- [       Thread-2] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : cb8625948210849f 2023-12-20 16:29:31.518  INFO 30856 --- [       Thread-2] o.a.k.c.c.internals.AbstractCoordinator  : Discovered coordinator 192.168.56.105:9092 (id: 2147483647 rack: null) for group thetechcheck. 2023-12-20 16:29:31.521  INFO 30856 --- [       Thread-2] o.a.k.c.c.internals.ConsumerCoordinator  : Revoking previously assigned partitions [] for group thetechcheck 2023-12-20 16:29:31.522  INFO 30856 --- [       Thread-2] o.a.k.c.c.internals.AbstractCoordinator  : (Re-)joining group thetechcheck 2023-12-20 16:29:31.564  INFO 30856 --- [       Thread-2] o.a.k.c.c.internals.AbstractCoordinator  : Successfully joined group thetechcheck with generation 3 2023-12-20 16:29:31.567  INFO 30856 --- [       Thread-2] o.a.k.c.c.internals.ConsumerCoordinator  : Setting newly assigned partitions [thetechcheck-0] for group thetechcheck 2023-12-20 16:29:31.618  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Received message: The index is now: 0 2023-12-20 16:29:31.619 ERROR 30856 --- [       Thread-2] c.c.p.S..k.SimpleKafkaConsumer            : A JSONObject text must begin with '{' at character 1 of The index is now: 0 2023-12-20 16:29:31.630  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Offset committed to Kafka. 2023-12-20 16:29:31.631  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Received message: The index is now: 2023-12-20 16:29:31.619 ERROR 30856 --- [       Thread-2] c.c.p.S..k.SimpleKafkaConsumer            : A JSONObject text must begin with '{' at character 1 of The index is now: 0 2023-12-20 16:29:31.630  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Offset committed to Kafka. // AP에서 받은 메시지 ↓ 2023-12-20 16:29:31.631  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Received message: The index is now: 1 2023-12-20 16:29:31.631 ERROR 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : A JSONObject text must begin with '{' at character 1 of The index is now: 1 2023-12-20 16:29:31.636  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Offset committed to Kafka. 2023-12-20 16:29:31.636  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Received message: The index is now: 2 2023-12-20 16:29:31.636 ERROR 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : A JSONObject text must begin with '{' at character 1 of The index is now: 2 2023-12-20 16:29:31.640  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Offset committed to Kafka. 2023-12-20 16:29:31.641  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Received message: The index is now: 3 2023-12-20 16:29:31.641 ERROR 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : A JSONObject text must begin with '{' at character 1 of The index is now: 3 2023-12-20 16:29:31.646  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Offset committed to Kafka. 2023-12-20 16:29:31.646  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Received message: The index is now: 4 2023-12-20 16:29:31.646 ERROR 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : A JSONObject text must begin with '{' at character 1 of The index is now: 4 2023-12-20 16:29:31.653  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Offset committed to Kafka. 2023-12-20 16:29:31.654  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Received message: The index is now: 5 2023-12-20 16:29:31.654 ERROR 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : A JSONObject text must begin with '{' at character 1 of The index is now: 5 2023-12-20 16:29:31.661  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Offset committed to Kafka. 2023-12-20 16:29:31.661  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Received message: The index is now: 6 2023-12-20 16:29:31.661 ERROR 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : A JSONObject text must begin with '{' at character 1 of The index is now: 6 2023-12-20 16:29:31.667  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Offset committed to Kafka. 2023-12-20 16:29:31.667  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Received message: The index is now: 7 2023-12-20 16:29:31.668 ERROR 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : A JSONObject text must begin with '{' at character 1 of The index is now: 7 2023-12-20 16:29:31.672  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Offset committed to Kafka. 2023-12-20 16:29:31.673  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Received message: The index is now: 8 2023-12-20 16:29:31.673 ERROR 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : A JSONObject text must begin with '{' at character 1 of The index is now: 8 2023-12-20 16:29:31.677  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Offset committed to Kafka. 2023-12-20 16:29:31.678  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Received message: The index is now: 9 2023-12-20 16:29:31.678 ERROR 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : A JSONObject text must begin with '{' at character 1 of The index is now: 9 2023-12-20 16:29:31.683  INFO 30856 --- [       Thread-2] c.c.p.S.k.SimpleKafkaConsumer            : Offset committed to Kafka.