Hi Flink team. Please can you help me with this error. Can you see why I would be getting this error in my application?

Caused by: java.lang.ClassCastException: class io.trurecognition.platform.model.PersonCount cannot be cast to class io.trurecognition.platform.model.Person (io.trurecognition.platform.model.PersonCount and io.trurecognition.platform.model.Person are in unnamed module of loader org.apache.flink.util.ChildFirstClassLoader @380c0ed3)

I can see the data is correctly formed on my Kafka topics and there definitely isn't a mix up of data on there. In my main method I'm setting it up like this:

env = StreamExecutionEnvironment.getExecutionEnvironment();
kafkaConnector = new KafkaConnector();

FlinkKafkaConsumer<String> sourcePeople = getKafkaConsumer("domain_person");
FlinkKafkaConsumer<String> sourceCount = getKafkaConsumer("domain_people");
FlinkKafkaProducer<String> sink = getKafkaProducer("derived_event");

DataStream<Person> personStream = createPersonStream(sourcePeople);
DataStream<PersonCount> countStream = createCountStream(sourceCount);

DataStream<String> combinedStream = connectAndProcessStreams(personStream, countStream);

Here are my flink data streams.

/**
   * Creates a Flink DataStream of Person objects from the given Kafka source. Invalid data that
   * cannot be parsed to a Person object is discarded.
   *
   * @param source Kafka consumer
   * @return DataStream of Person objects
   */
  private static DataStream<Person> createPersonStream(FlinkKafkaConsumer<String> source) {
    return env.addSource(source)
        .map(
            value -> {
              try {
                return getObjectMapper().readValue(value, Person.class);
              } catch (Exception e) {
                LOG.error("Error parsing Person: " + value, e);
                return null;
              }
            })
        .filter(Objects::nonNull); // Filter out null values
  }

  /**
   * Creates a Flink DataStream of PersonCount objects from the given Kafka source. Invalid data
   * that cannot be parsed to a PersonCount object is discarded.
   *
   * @param source Kafka consumer
   * @return DataStream of PersonCount objects
   */
  private static DataStream<PersonCount> createCountStream(FlinkKafkaConsumer<String> source) {
    return env.addSource(source)
        .map(
            value -> {
              try {
                return getObjectMapper().readValue(value, PersonCount.class);
              } catch (Exception e) {
                LOG.error("Error parsing PersonCount: " + value, e);
                return null;
              }
            })
        .filter(Objects::nonNull); // Filter out null values
  }

  /**
   * Connects two data streams and processes them using a CoFlatMapFunction. The processing involves
   * updating a TemporalCache with data from both streams, and emitting a message if a certain
   * condition is met.
   *
   * @param personStream DataStream of Person objects
   * @param countStream DataStream of PersonCount objects
   * @return A combined data stream of Strings TODO: make private
   */
  public static DataStream<String> connectAndProcessStreams(
      DataStream<Person> personStream, DataStream<PersonCount> countStream) {
    DataStream<CacheUpdateResult> personResultStream =
        personStream.map(new PersonToCacheUpdateResultFunction()).returns(CacheUpdateResult.class);

    DataStream<CacheUpdateResult> countResultStream =
        countStream.map(new CountToCacheUpdateResultFunction()).returns(CacheUpdateResult.class);

    return personResultStream
        .union(countResultStream)
        .flatMap(new CacheUpdateResultToStringFunction())
        .returns(String.class);
  }

We're using a temporal cache to perform certain calculations. Let me know if I need to share more code for you to diagnose the issue.