Wrong Serdes for different binders

[fabiofilz]

Hi,

We are migrating from AVRO to Proto so we have some topics using Avro and other Proto.

I am trying to use a function that has 2 parameters as input: the first one is AVRO and second is Proto and the output is Avro.

When I try to run the application tries to deserialise the Avro event using Proto Serdes.

Please find below the configuration:

`

application.yml

spring:
cloud:
stream:
function:
definition: processTopicEventProto;processTopicEventAvro

  binders:
    kstream-proto:
      type: kstream
      environment:
        spring.cloud.stream.kafka.streams.binder:
          brokers: localhost:9092
          configuration:
            default.value.serde: io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde
            schema.registry.url: http://localhost:8081
            derive.type: true

    kstream-avro:
      type: kstream
      environment:
        spring.cloud.stream.kafka.streams.binder:
          brokers: localhost:9092
          configuration:
            specific.avro.reader: true
            default.value.serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
            schema.registry.url: http://localhost:8081

    globalktable:
      type: globalktable
      environment:
        spring.cloud.stream.kafka.streams.binder:
          brokers: localhost:9092
          configuration:
            default.value.serde: io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde
            schema.registry.url: http://localhost:8081
            derive.type: true

    processTopicEventProto-in-0:
      destination: topicEvent.proto
      binder: kstream-proto
      contentType: application/x-protobuf
    processTopicEventProto-in-1:
      destination: consent.proto
      binder: globalktable
      contentType: application/x-protobuf
    processTopicEventProto-out-0:
      destination: filteredTopicEvent.proto
      binder: kstream-proto
      contentType: application/x-protobuf

    processTopicEventAvro-in-0:
      destination: topicEvent.avro
      binder: kstream-avro
    processTopicEventAvro-in-1:
      destination: consent.proto
      binder: globalktable
    processTopicEventAvro-out-0:
      destination: filteredTopicEvent.avro
      binder: kstream-avro

  kafka:
    streams:
      binder:
        deserialization-exception-handler: sendtodlq
        functions:
          processTopicEventProto.applicationId: service-processTopicEventProto
          processTopicEventAvro.applicationId: service-processTopicEventAvro
        configuration:
          state.dir: state-store
          acceptable.recovery.lag: 0

Topology

@Bean
public BiFunction<KStream<String, SpecificRecord>, GlobalKTable<String, CustomerConsentProto.CustomerConsent>, KStream<String, SpecificRecord>> processTopicEventAvro() {

    return (stream, table) -> (
        stream
            .peek((key, value) -> log.info("Received event key={} value={}", key, value.toString()))
            .leftJoin(table, (key, value) -> key, (streamRecord, tableRecord) -> {
                if (tableRecord == null){
                    return streamRecord;
                }
                return null;
            } )
            .filter((key, value) -> Optional.ofNullable(value).isPresent())
            .peek((key, value) -> log.info("The Event customerId={} will go to the output", key))
    );
}


@Bean
public BiFunction<KStream<String, com.google.protobuf.Message>, GlobalKTable<String, CustomerConsentProto.CustomerConsent>, KStream<String, com.google.protobuf.Message>> processTopicEventProto() {

    return (stream, table) -> (
            stream
                .peek((key, value) -> log.debug("Received event key={} value={}", key, value.toString()))
                .leftJoin(table, (key, value) -> key, (streamRecord, tableRecord) -> {
                    if (tableRecord == null){
                        return streamRecord;
                    }
                    return null;
                } )
                .filter((key, value) -> Optional.ofNullable(value).isPresent())
                .peek((key, value) -> log.info("The Event customerId={} will go to the output", key))
    );
}

Logs

2022-10-25 15:28:53.568  WARN 44031 --- [-StreamThread-1] o.a.k.s.p.internals.RecordDeserializer   : stream-thread [service-processTopicEventAvro-69ad1666-13ca-4550-8dd4-8a6fc65293d3-StreamThread-1] task [0_2] Skipping record due to deserialization error. topic=[topicEvent.avro] partition=[2] offset=[2669629]

org.apache.kafka.common.errors.SerializationException: Error deserializing Protobuf message for id 744
at io.confluent.kafka.serializers.protobuf.AbstractKafkaProtobufDeserializer.deserialize(AbstractKafkaProtobufDeserializer.java:175) ~[kafka-protobuf-serializer-7.2.1.jar:na]
at io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer.deserialize(KafkaProtobufDeserializer.java:76) ~[kafka-protobuf-serializer-7.2.1.jar:na]
at io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer.deserialize(KafkaProtobufDeserializer.java:27) ~[kafka-protobuf-serializer-7.2.1.jar:na]
at org.apache.kafka.common.serialization.Deserializer.deserialize(Deserializer.java:60) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:58) ~[kafka-streams-3.1.1.jar:na]
at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:66) ~[kafka-streams-3.1.1.jar:na]
at org.apache.kafka.streams.processor.internals.RecordQueue.updateHead(RecordQueue.java:176) ~[kafka-streams-3.1.1.jar:na]
at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:112) ~[kafka-streams-3.1.1.jar:na]
at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:304) ~[kafka-streams-3.1.1.jar:na]
at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:960) ~[kafka-streams-3.1.1.jar:na]
at org.apache.kafka.streams.processor.internals.TaskManager.addRecordsToTasks(TaskManager.java:1068) ~[kafka-streams-3.1.1.jar:na]
at org.apache.kafka.streams.processor.internals.StreamThread.pollPhase(StreamThread.java:962) ~[kafka-streams-3.1.1.jar:na]
at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:751) ~[kafka-streams-3.1.1.jar:na]
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:604) ~[kafka-streams-3.1.1.jar:na]
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:576) ~[kafka-streams-3.1.1.jar:na]
Caused by: java.io.IOException: Invalid schema << IT SHOWS THE AVRO SCHEMA HERE >> with refs [] of type AVRO
at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.lambda$getSchemaByIdFromRegistry$6(CachedSchemaRegistryClient.java:305) ~[kafka-schema-registry-client-7.2.1.jar:na]
at java.base/java.util.Optional.orElseThrow(Optional.java:403) ~[na:na]
at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaByIdFromRegistry(CachedSchemaRegistryClient.java:303) ~[kafka-schema-registry-client-7.2.1.jar:na]
at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaBySubjectAndId(CachedSchemaRegistryClient.java:417) ~[kafka-schema-registry-client-7.2.1.jar:na]
at io.confluent.kafka.serializers.protobuf.AbstractKafkaProtobufDeserializer.deserialize(AbstractKafkaProtobufDeserializer.java:126) ~[kafka-protobuf-serializer-7.2.1.jar:na]
... 14 common frames omitted

2022-10-25 15:28:53.814 ERROR 44031 --- [-StreamThread-1] i.c.k.s.c.CachedSchemaRegistryClient : Invalid schema type AVRO

Libs

JAVA 17

org.springframework.boot
spring-boot-starter-parent
2.7.5

<spring-cloud.version>2021.0.4</spring-cloud.version>

`

[sobychacko]

@fabiofilz I moved the issue to core Spring Cloud Stream as this is the new home for Kafka binders.

When writing an app that uses Kafka Streams binder only, then it is recommended to not use a multi-binder setup. I wonder if that is the root of the problems you are running into or not.

Could you try the following configuration instead and see if you notice any difference?

spring.cloud.streams.kafka.streams:
  binder:
    deserialization-exception-handler: sendtodlq
    functions:
      processTopicEventProto.applicationId: service-processTopicEventProto
      processTopicEventAvro.applicationId: service-processTopicEventAvro
    configuration:
      state.dir: state-store
      acceptable.recovery.lag: 0
      schema.registry.url: http://localhost:8081
      specific.avro.reader: true
      derive.type: true
  bindings:
    processTopicEventAvro-in-0:
      destination: topicEvent.avro
      consumer:
        valueSerde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
    processTopicEventAvro-in-1:
      destination: consent.proto
      consumer:
        valueSerde: io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde
    processTopicEventAvro-out-0:
      destination: filteredTopicEvent.avro
      consumer:
        valueSerde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
    processTopicEventProto-in-0:
      destination: topicEvent.proto
      consumer:
        valueSerde: io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde
    processTopicEventProto-in-1:
      destination: consent.proto
      consumer:
        valueSerde: io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde
    processTopicEventProto-out-0:
      destination: filteredTopicEvent.proto
      consumer:
        valueSerde: io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde

Since you have multiple input bindings, we can't really provide a default.value.serde through configuration - only per binding. You need to make sure that anywhere in your application logic where it expects a Serde, you need to call the right overloaded method that provides the Serde since there is no default.value.serde provided and the default used by Kafka Streams (byte[]) will not work in this case since you have AVRO and Protobuf.

If you still face issues, please create a small simple application where we can reproduce the issue standalone. Then we can triage further. Thanks!

[fabiofilz]

Hi @sobychacko,

Thank you for your quick response.

The second parameter is a globalktable binder, but I don't see it in your example.

I left the second input parameter as it is and changed the first (input) and third (output), but it didn't work. It shows the same error.

Thank you.
Fabio

[sobychacko]

@fabiofilz There is nothing specific about GlobalKTable binding from a configuration standpoint. The binder auto-detects those things. Could you create a small sample app and share with us? Please make sure that we can run and reproduce the issue standalone (provide a README maybe?). Then we will be happy to help further. Thanks!

[sobychacko]

@fabiofilz Any updates on this issue?

[fabiofilz]

Hi @sobychacko,

Sorry for the late response, but I had to create an app to simulate the error:

https://github.com/fabiofilz/spring-cloud-kafka-streams-stream-globalktable

I added the information on the readme file.

Please let me know if you need further information.

Thank you.
Fabio

[sobychacko]

@fabiofilz Could you please break your sample app into a simple standalone Boot app with minimal configurations? As I indicated, please remove all the multi-binder setups in the configuration which is not needed for Kafka Streams applications. We need a simple sample where we can run the app and get the original serialization errors you reported here.

[fabiofilz]

Hi @sobychacko,
I created a simpler example as you suggested here https://github.com/fabiofilz/spring-cloud-kafka-streams-stream-globalktable-v3
I added instructions on how to simulate the error when the first parameter of the function is avro and the second is proto (GlobalKTable). I also added another example when the first parameter of the function is proto and the second is proto (GlobalKTable) and it works fine.
Please let me know if it is clear now.
Thank you

[sobychacko]

@fabiofilz I cannot build your sample app from the root of your repository (for example, mvn clean package fails). It would help if you had the confluent repo definition in the pom.xml. Can you make sure that the project is buildable? Please remove the individual binder configurations here and move them as individual binding configurations. As far as I can see, you do not have a multi-binder use case, so separate binder configurations are unnecessary. Also, please remove all the native encoding/decoding and content-type configurations. For the Kafka Streams binder, we recommend relying on native decoding (which is true by default). Once you have your config cleaned up with all these changes, paste it on this issue, and we can review it further. Also, feel free to update the sample app at that point.

[fabiofilz]

Hi @sobychacko,
Thank you for your response.

1 - What do you mean by "confluent repo definition in the pom.xml", please?

2 - I would like to double-check about the binders. I checked this link https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream-binder-kafka.html#_kafka_streams_binder:~:text=0.destination%3Dinput.*-,Multiple%20Input%20Bindings,-Many%20non%2Dtrivial

The binder kstream-avro has AVRO serde, stream has Proto serdes and globalktable has Proto serdes. How can I specify the serdes on binding?

3 - Could you please paste the building error? I can execute "mvn clean package"
PS.: The example doesn't have Junit and the test is done manually as per README file

[INFO] --- maven-jar-plugin:3.2.2:jar (default-jar) @ spring-cloud-kafka-streams-stream-globalktable-application ---
[INFO] Building jar: /Users/fabsim/git/courses/spring-cloud-kafka-streams-stream-globalktable-v3/application/target/spring-cloud-kafka-streams-stream-globalktable-application-1.0-SNAPSHOT.jar
[INFO]
[INFO] --- spring-boot-maven-plugin:2.7.5:repackage (repackage) @ spring-cloud-kafka-streams-stream-globalktable-application ---
[INFO] Replacing main artifact with repackaged archive
[INFO] ------------------------------------------------------------------------
[INFO] Reactor Summary for spring-cloud-kafka-streams-stream-globalktable 1.0-SNAPSHOT:
[INFO]
[INFO] spring-cloud-kafka-streams-stream-globalktable ..... SUCCESS [ 0.166 s]
[INFO] spring-cloud-kafka-streams-stream-globalktable-messages SUCCESS [ 4.945 s]
[INFO] spring-cloud-kafka-streams-stream-globalktable-application SUCCESS [ 8.298 s]
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 14.472 s
[INFO] Finished at: 2023-01-25T07:01:18-03:00
[INFO] ------------------------------------------------------------------------

Thanks

[sobychacko]

@fabiofilz You probably already have the artifacts locally cached. I get errors as below.

[ERROR] Failed to execute goal org.xolstice.maven.plugins:protobuf-maven-plugin:0.6.1:compile (default) on project spring-cloud-kafka-streams-stream-globalktable-messages: Unable to resolve artifact: Missing:
[ERROR] ----------
[ERROR] 1) com.google.protobuf:protoc:exe:osx-aarch_64:3.13.0
[ERROR]
[ERROR]   Try downloading the file manually from the project website.
[ERROR]
[ERROR]   Then, install it using the command:
[ERROR]       mvn install:install-file -DgroupId=com.google.protobuf -DartifactId=protoc -Dversion=3.13.0 -Dclassifier=osx-aarch_64 -Dpackaging=exe -Dfile=/path/to/file
[ERROR]
[ERROR]   Alternatively, if you host your own repository you can deploy the file there:
[ERROR]       mvn deploy:deploy-file -DgroupId=com.google.protobuf -DartifactId=protoc -Dversion=3.13.0 -Dclassifier=osx-aarch_64 -Dpackaging=exe -Dfile=/path/to/file -Durl=[url] -DrepositoryId=[id]
[ERROR]
[ERROR]   Path to dependency:
[ERROR]   	1) com.fabio.springcloudexample:spring-cloud-kafka-streams-stream-globalktable-messages:jar:1.0-SNAPSHOT
[ERROR]   	2) com.google.protobuf:protoc:exe:osx-aarch_64:3.13.0
[ERROR]
[ERROR] ----------
[ERROR] 1 required artifact is missing.

You need to add the Confluent package repo as a maven repository in the pom. I added it, but still got the above error. Try building with -Dmaven.repo.local=<some-directory> to a non-standard m2 location (not the standard m2). Then, you should get the error.

[sobychacko]

I tried to simplify your config as below. I haven't tried the code yet, but please compare it with yours and see if you can run it. I took out all the multi-binder configs, native decoding/encoding configs, etc.

spring:
  cloud:
    stream:
      function:
        definition: processProto;processAvro
      bindings:
        processProto-in-0:
          destination: OrdersInput-proto
        processProto-in-1:
          destination: EnabledCustomer-proto
        processProto-out-0:
          destination: OrdersOutput-proto
        processAvro-in-0:
          destination: OrdersInput-avro
        processAvro-in-1:
          destination: EnabledCustomer-proto
        processAvro-out-0:
          destination: OrdersOutput-avro
      kafka:
        streams:
          binder:
            deserialization-exception-handler: sendtodlq
            functions:
              processProto.applicationId: spring-cloud-kafka-streams-stream-globalktable-processProto
              processAvro.applicationId: spring-cloud-kafka-streams-stream-globalktable-processAvro
            configuration:
              state.dir: state-store
              acceptable.recovery.lag: 0
              schema.registry.url: http://localhost:8081
          bindings:
            processProto-in-0:
             consumer:
               configuration:
                 default.value.serde: io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde
                 derive.type: true
            processProto-in-1:
              consumer:
                configuration:
                  default.value.serde: io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde
                  derive.type: true
            processProto-out-0:
              consumer:
                configuration:
                  default.value.serde: io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde
                  derive.type: true
            processAvro-in-0:
              consumer:
                configuration:
                  specific.avro.reader: true
                  default.key.serde: io.confluent.kafka.streams.serdes.avro.PrimitiveAvroSerde
                  default.value.serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
            processAvro-in-1:
              consumer:
                configuration:
                  default.value.serde: io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde
                  derive.type: true
            processAvro-out-0:
              consumer:
                configuration:
                  specific.avro.reader: true
                  default.key.serde: io.confluent.kafka.streams.serdes.avro.PrimitiveAvroSerde
                  default.value.serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
        binder:
          configuration:
            heartbeat.interval.ms: 10000
            max.poll.records: 5000

[fabiofilz]

Hi @sobychacko,

1 - Building error
1.1 - I added the repository https://packages.confluent.io/maven/
1.2 - I renamed my .m2 folder and execute "mvn clean package -Dmaven.repo.local=~/xxx"
PS.: I am using Mac Intel I9

[INFO] Replacing main artifact with repackaged archive
[INFO] ------------------------------------------------------------------------
[INFO] Reactor Summary for spring-cloud-kafka-streams-stream-globalktable 1.0-SNAPSHOT:
[INFO]
[INFO] spring-cloud-kafka-streams-stream-globalktable ..... SUCCESS [ 0.281 s]
[INFO] spring-cloud-kafka-streams-stream-globalktable-messages SUCCESS [ 7.973 s]
[INFO] spring-cloud-kafka-streams-stream-globalktable-application SUCCESS [01:58 min]
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 02:07 min
[INFO] Finished at: 2023-01-25T15:09:47-03:00
[INFO] ------------------------------------------------------------------------

2 - Deserialization error:
2.1 - Deleted the folder state-store
2.2 - Replaced application.yml with your configuration
2.3 - Run the application
2.4 - Executed application/src/test/java/kafka/producer/CustomerEnabled.java
2.5 - I can see the message produced into the topic EnabledCustomer-proto using AKHQ http://localhost:8080/

---

org.apache.kafka.common.errors.SerializationException: Can't deserialize data [[0, 0, 0, 0, 1, 0, 10, 6, 49, 50, 51, 52, 53, 54]] from topic [EnabledCustomer-proto]
at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:588) ~[spring-kafka-2.8.10.jar:2.8.10]
at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:58) ~[kafka-streams-3.1.2.jar:na]
at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:66) ~[kafka-streams-3.1.2.jar:na]
at org.apache.kafka.streams.processor.internals.GlobalStateUpdateTask.update(GlobalStateUpdateTask.java:96) ~[kafka-streams-3.1.2.jar:na]
at org.apache.kafka.streams.processor.internals.GlobalStreamThread$StateConsumer.pollAndUpdate(GlobalStreamThread.java:262) ~[kafka-streams-3.1.2.jar:na]
at org.apache.kafka.streams.processor.internals.GlobalStreamThread.run(GlobalStreamThread.java:312) ~[kafka-streams-3.1.2.jar:na]
Caused by: java.io.CharConversionException: Invalid UTF-32 character 0xff0a06 (above 0x0010ffff) at char #1, byte #7)
at com.fasterxml.jackson.core.io.UTF32Reader.reportInvalid(UTF32Reader.java:193) ~[jackson-core-2.13.4.jar:2.13.4]
at com.fasterxml.jackson.core.io.UTF32Reader.read(UTF32Reader.java:159) ~[jackson-core-2.13.4.jar:2.13.4]
at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._loadMore(ReaderBasedJsonParser.java:276) ~[jackson-core-2.13.4.jar:2.13.4]
at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._skipWSOrEnd(ReaderBasedJsonParser.java:2442) ~[jackson-core-2.13.4.jar:2.13.4]
at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:698) ~[jackson-core-2.13.4.jar:2.13.4]
at com.fasterxml.jackson.databind.ObjectReader._initForReading(ObjectReader.java:356) ~[jackson-databind-2.13.4.2.jar:2.13.4.2]
at com.fasterxml.jackson.databind.ObjectReader._bindAndClose(ObjectReader.java:2041) ~[jackson-databind-2.13.4.2.jar:2.13.4.2]
at com.fasterxml.jackson.databind.ObjectReader.readValue(ObjectReader.java:1529) ~[jackson-databind-2.13.4.2.jar:2.13.4.2]
at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:585) ~[spring-kafka-2.8.10.jar:2.8.10]
... 5 common frames omitted

2023-01-25 15:16:51.320 WARN 280 --- [balStreamThread] o.a.k.s.p.internals.RecordDeserializer : global-stream-thread [spring-cloud-kafka-streams-stream-globalktable-processProto-e9c7ee99-bc21-4f4d-a639-ff4f4ed47f0f-GlobalStreamThread] Skipping record due to deserialization error. topic=[EnabledCustomer-proto] partition=[0] offset=[0]

org.apache.kafka.common.errors.SerializationException: Can't deserialize data [[0, 0, 0, 0, 1, 0, 10, 6, 49, 50, 51, 52, 53, 54]] from topic [EnabledCustomer-proto]
at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:588) ~[spring-kafka-2.8.10.jar:2.8.10]
at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:58) ~[kafka-streams-3.1.2.jar:na]
at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:66) ~[kafka-streams-3.1.2.jar:na]
at org.apache.kafka.streams.processor.internals.GlobalStateUpdateTask.update(GlobalStateUpdateTask.java:96) ~[kafka-streams-3.1.2.jar:na]
at org.apache.kafka.streams.processor.internals.GlobalStreamThread$StateConsumer.pollAndUpdate(GlobalStreamThread.java:262) ~[kafka-streams-3.1.2.jar:na]
at org.apache.kafka.streams.processor.internals.GlobalStreamThread.run(GlobalStreamThread.java:312) ~[kafka-streams-3.1.2.jar:na]
Caused by: java.io.CharConversionException: Invalid UTF-32 character 0xff0a06 (above 0x0010ffff) at char #1, byte #7)
at com.fasterxml.jackson.core.io.UTF32Reader.reportInvalid(UTF32Reader.java:193) ~[jackson-core-2.13.4.jar:2.13.4]
at com.fasterxml.jackson.core.io.UTF32Reader.read(UTF32Reader.java:159) ~[jackson-core-2.13.4.jar:2.13.4]
at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._loadMore(ReaderBasedJsonParser.java:276) ~[jackson-core-2.13.4.jar:2.13.4]
at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._skipWSOrEnd(ReaderBasedJsonParser.java:2442) ~[jackson-core-2.13.4.jar:2.13.4]
at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:698) ~[jackson-core-2.13.4.jar:2.13.4]
at com.fasterxml.jackson.databind.ObjectReader._initForReading(ObjectReader.java:356) ~[jackson-databind-2.13.4.2.jar:2.13.4.2]
at com.fasterxml.jackson.databind.ObjectReader._bindAndClose(ObjectReader.java:2041) ~[jackson-databind-2.13.4.2.jar:2.13.4.2]
at com.fasterxml.jackson.databind.ObjectReader.readValue(ObjectReader.java:1529) ~[jackson-databind-2.13.4.2.jar:2.13.4.2]
at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:585) ~[spring-kafka-2.8.10.jar:2.8.10]
... 5 common frames omitted

[sobychacko]

@fabiofilz Is the sample app updated with all these changes?