[SPARK-50332] migrate kafka consumer offset information in spark to new MSK cluster

• Type: Sub-task
• Status: Open
• Priority: Blocker
• Resolution: Unresolved
• Affects Version/s: 4.0.0, 3.5.2
• Component/s: Connect, Tests
• Labels: pull-request-available
• Target Version/s: 4.0.0
• Assignee: Unassigned
• Reporter: Ramakrishna
• Created: 3 days ago 05:25
• Updated: Yesterday 06:39

Description

I have spark job that reads messages from kafka , using

 

            

 

```spark.readStream(.....)```

 

method. This is deployed in AWS as sparkapp and it reads data from MSK kafka. Recently the devOPs decided to migrate to a different MSK cluster for cost reduction.

There were some go Lang services which use their kafka offsets maintained by kafka server itself and the all the consumer offsets were automaticallly migrated to new MSK cluster since it is being managed by kafka itself.

However for spark jobs we have structured streaming which has checkpointing which is being managed by spark job itself. So when this spark job is being made to listen to new kafka URLs , they fail with missing offsets error. To resolve this we need to set failOnDataLoss = false , or delete checkpoint, in either case we have data loss and state gets lost, if the job has some custom state logic .

I have 2 requirements

1. Is there a way that kafka offsets are migrated automatically for spark consumers too whenever there is a migration of MSK, assuming that spark job has structured streaming.

      2. Is there a way that we can delete checkpoint, so that we can remove the offsets tracked by checkpoint, in order to reprocess with "earliest" offset, however without losing the Spark state .
 

Issue Links

• is a clone of: SPARK-48139 Re-enable `SparkSessionE2ESuite.interrupt tag`
• links to: GitHub Pull Request #48896