Kafka Consumer not consuming reprocessed data with Spark

We're using a pyspark application in order to process some data from a source topic in Kafka and write the processed data in a separate topic. We've got another python application which uses Kafka-python to consume from the processed topic. Everything goes correct the first run.

Later we decide we want to add another column to the processed topic. The source topic already contains this information so we stop the old Kafka/Spark stream, start a new one which does exactly the same except it includes the new column. The new stream is started with startingOffsets set to earliest The new column is printed to the console so I assume the column is now included in the processed topic.

The python application with the consumer, stops the old consumer and starts a new one as well. This one is started with startingOffsets set to latest. The issue is the consumer doesn't consume the new processed data it seems. Somehow the reprocessed data isn't triggering a consume with the python application. Am I missing something here?

BTW: when using startingOffsets set to earliest on the python application with consumer, it starts consuming all the old data, but not the newly processed data with the new column.

Example:

First run

Source topic contains the following:

|column1|column2|column3|column4|column5|column6|

This is processed by the pyspark application to the processed topic:

|column1|column2|column4|column6|

And the python with consumer application consumes it. Even if an hour later new data comes in, it's processed.

After update run

Source topic contains the following:

|column1|column2|column3|column4|column5|column6|

This is processed by the pyspark application to the processed topic:

|column1|column2|column4|column5|column6|

This isn't consumed by the python application.

Update

Code from pyspark app:

self.data_frame = self.spark_session.readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", KAFKA_HOST) \
            .option("subscribe", self.compute_config.topic_sources[0]) \
            .option("startingOffsets", "earliest" if reset_offset is True else "latest") \
            .option("failOnDataLoss", False) \
            .load()

...

self.ds = self.data_frame.select("key", from_json(col("value").cast("string"), self.schema).alias("value")) \
            .withColumn("value", col("value").cast(self.rename_schema)) \
            .withColumn("value", to_json("value")) \
            .writeStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", KAFKA_HOST) \
            .option("topic", self.compute_config.topic_target) \
            .option("checkpointLocation", f"/app/checkpoints/{self.compute_config.topic_target+str(self.compute_config.id)}") \
            .start()

Code from python app with consumer:

 self.consumer = KafkaConsumer(self.sink_config.topic_target,
                                      bootstrap_servers=[KAFKA_HOST],
                                      auto_offset_reset="latest",
                                      enable_auto_commit=False)

for message in self.consumer:
    ...