KStream-KStream inner join throws java.lang.ClassCastException

I downloaded your code from GitHub to dig into this, and turns out it's actually a bug in the used JsonSerializer/JsonDeserializer. The type (School, Person, PersonKey, SchoolKey) is encoded in the record headers, but the headers are never cleaned up. Each time the type changes, just a new header is appended (header keys are not unique, and duplicates are allowed).

For some record, the same type is just encoded multiple times, and thus, this part of the code works. However, for some cases, different types are encoded and one type (the first header found) is picked "randomly" when reading data from a topic. The happens before the join, but when receiving data from a repartition topic. If the wrong type is picked, the code crashed with a ClassCastException later.

New Answer:

Following the discussion on this ticket, https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/685, you should disable that type information is written into the record headers via:

props.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);

Note, that all Serdes that are created manually, ie, via calling new must be configured manually:

Map<String, Object> config = new HashMap<>();
config.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);

PersonKeySerde personKeySerde = new PersonKeySerde();
personKeySerde.configure(config, true);

PersonSerde personSerde = new PersonSerde();
personSerde.configure(config, false);

// ...
.through("person", Produced.with(personKeySerde, personSerde));

Original Answer:

As a workaround, you can replace map and selectKey() with a transform() and clear the headers within transform(). It's a hack though. You should file a ticket agains SpringBoot project so they can fix JsonSerializer/JsonDeserializer.

The following code removes the headers and ensure that the correct types are used, avoiding the ClassCastException:

@SpringBootApplication
public class KafkaStreamJoinApplication {

    public static void main(String[] args) {
        SpringApplication.run(KafkaStreamJoinApplication.class, args);
    }

    @EnableBinding(KStreamProcessorX.class)
    public static class KafkaKStreamJoinApplication {

        @StreamListener
        public void process(@Input("school") KStream<SchoolKey, School> schools) {
            // replace map() with transform()
            schools.transform(new TransformerSupplier<SchoolKey, School, KeyValue<PersonKey, Person>>() {
                @Override
                public Transformer<SchoolKey, School, KeyValue<PersonKey, Person>> get() {
                    return new Transformer<SchoolKey, School, KeyValue<PersonKey, Person>>() {
                        ProcessorContext context;

                        @Override
                        public void init(final ProcessorContext context) {
                            this.context = context;
                        }

                        @Override
                        public KeyValue<PersonKey, Person> transform(final SchoolKey key, final School value) {
                            // clear all headers; would be sufficient to only remove type header
                            for (Header h : context.headers().toArray()) {
                                context.headers().remove(h.key());
                            }
                            // same a "old" map code:
                            return KeyValue.pair(new PersonKey("Adam", "Smith", key.getId()), new Person(12));
                        }

                        @Override
                        public void close() {}
                    };
                }})
                .through("person", Produced.with(new PersonKeySerde(), new PersonSerde()));
        }

        @StreamListener
        public void process1(@Input("school_1") KStream<SchoolKey, School> schools, @Input("person") KStream<PersonKey, Person> persons) {

            // replace selectKey() with transform()
            schools.transform(new TransformerSupplier<SchoolKey, School, KeyValue<Integer, School>>() {
                @Override
                public Transformer<SchoolKey, School, KeyValue<Integer, School>> get() {
                    return new Transformer<SchoolKey, School, KeyValue<Integer, School>>() {
                        ProcessorContext context;

                        @Override
                        public void init(final ProcessorContext context) {
                            this.context = context;
                        }

                        @Override
                        public KeyValue<Integer, School> transform(final SchoolKey key, final School value) {
                            // clear all headers; would be sufficient to only remove type header
                            for (Header h : context.headers().toArray()) {
                                context.headers().remove(h.key());
                            }
                            // effectively the same as "old" selectKey code:
                            return KeyValue.pair(key.getId(), value);
                        }

                        @Override
                        public void close() {}
                    };
                }})
                // replace selectKey() with transform()
                .join(persons.transform(new TransformerSupplier<PersonKey, Person, KeyValue<Integer, Person>>() {
                    @Override
                    public Transformer<PersonKey, Person, KeyValue<Integer, Person>> get() {
                        return new Transformer<PersonKey, Person, KeyValue<Integer, Person>>() {
                            ProcessorContext context;

                            @Override
                            public void init(final ProcessorContext context) {
                                this.context = context;
                            }

                            @Override
                            public KeyValue<Integer, Person> transform(final PersonKey key, final Person value) {
                                // clear all headers; would be sufficient to only remove type header
                                for (Header h : context.headers().toArray()) {
                                    context.headers().remove(h.key());
                                }
                                // effectively same as "old" selectKey code:
                                return KeyValue.pair(key.getId(), value);
                            }

                            @Override
                            public void close() {}
                        };
                    }}),
                    (school, person) -> {
                        System.out.println("school_app2= " + school + ", person_app2= " + person);
                        return null;
                    },
                    JoinWindows.of(Duration.ofSeconds(1)),
                    Joined.with(Serdes.Integer(), new SchoolSerde(), new PersonSerde())
                );
        }
    }

    interface KStreamProcessorX {
        @Input("person")
        KStream<?, ?> inputPersonKStream();

        @Input("school")
        KStream<?, ?> inputSchoolKStream();

        @Input("school_1")
        KStream<?, ?> inputSchool1KStream();
    }
}

Could it be that there are some stale data somewhere in the topics or in the underlying changelog topics? Can you try to use new topics and different application-id's to see if it fixes your issues?

Here is a sample configuration to use:

spring.cloud.stream.bindings.school:
  destination: school-abc
spring.cloud.stream.kafka.streams.bindings.school:
  consumer:
    keySerde: kafka.streams.serde.SchoolKeySerde
    valueSerde: kafka.streams.serde.SchoolSerde
    application-id: stream-join-sample_diff_id_1

spring.cloud.stream.bindings.person:
  destination: person-abc
spring.cloud.stream.kafka.streams.bindings.person:
  consumer:
    keySerde: kafka.streams.serde.PersonKeySerde
    valueSerde: kafka.streams.serde.PersonSerde
    application-id: stream-join-sample_diff_id_2

spring.cloud.stream.bindings.school_1:
  destination: school-abc
spring.cloud.stream.kafka.streams.bindings.school_1:
  consumer:
    keySerde: kafka.streams.serde.SchoolKeySerde
    valueSerde: kafka.streams.serde.SchoolSerde
    application-id: stream-join-sample_diff_id_2

spring.cloud.stream.kafka.streams.binder:
  brokers: localhost
  configuration:
    default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
    default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
    commit.interval.ms: 100

Note that I changed the topic names, application id etc. You might want to update any producers that populate the topics.

Also, notice that you don't need to specify the content type, setting useNativeDecoding to false etc. since those are the defaults in the current version of the kafka streams binder.