Getting exception while using spark-sql-kafka-offset-committer
Version:

            <groupId>net.heartsavior.spark</groupId>
            <artifactId>spark-sql-kafka-offset-committer_2.12</artifactId>
            <version>0.4.0-spark-3.0</version>
        </dependency>

Error:
23/12/06 05:25:45 ERROR Utils: uncaught error in thread spark-listener-group-streams, stopping SparkContext
java.lang.NoSuchMethodError: 'org.apache.spark.sql.connector.read.InputPartition org.apache.spark.sql.execution.datasources.v2.DataSourceRDDPartition.inputPartition()'
at org.apache.spark.sql.kafka010.KafkaSourceInspector.$anonfun$extractSourceTopicsFromDataSourceV2$2(KafkaSourceInspector.scala:112)
at scala.collection.TraversableLike.$anonfun$flatMap$1(TraversableLike.scala:293)
at scala.collection.IndexedSeqOptimized.foreach(IndexedSeqOptimized.scala:36)
at scala.collection.IndexedSeqOptimized.foreach$(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:198)
at scala.collection.TraversableLike.flatMap(TraversableLike.scala:293)
at scala.collection.TraversableLike.flatMap$(TraversableLike.scala:290)
at scala.collection.mutable.ArrayOps$ofRef.flatMap(ArrayOps.scala:198)
at org.apache.spark.sql.kafka010.KafkaSourceInspector.$anonfun$extractSourceTopicsFromDataSourceV2$1(KafkaSourceInspector.scala:111)
at org.apache.spark.sql.kafka010.KafkaSourceInspector.$anonfun$extractSourceTopicsFromDataSourceV2$1$adapted(KafkaSourceInspector.scala:110)
at scala.collection.immutable.List.flatMap(List.scala:366)
at org.apache.spark.sql.kafka010.KafkaSourceInspector.extractSourceTopicsFromDataSourceV2(KafkaSourceInspector.scala:110)
at org.apache.spark.sql.kafka010.KafkaSourceInspector.$anonfun$populateKafkaParams$1(KafkaSourceInspector.scala:40)
at scala.collection.TraversableLike.$anonfun$flatMap$1(TraversableLike.scala:293)

When StreamingQuery is started but there are no new messages to process, this line throws a NullpointerException.

The reason is that lastExecution is null so exec will be Some(null).

The exception is thrown continuously until at least one message is processed.

Hi ,
We tried with your approach and able to commit the offsets manually with command "kafkaConsumer.commitSync()" . But when we try to read last committed offset through spark structured stream .option("startingOffsets","earliest"); it is not picking latest committed offsets .

At present we are using another kafka consumer to get last commited offsets of all partition of specific topic's by passing same groupId and passing those offsets to sprark like option("startingOffsets", "{"topicName":{"0":2764}}") to read the data . and its working as expected .

please let us know is there any way to get last committed offsets through spark structured stream.

As I got some amazing contribution on https://github.com/HeartSaVioR/spark-state-tools to support cross build, I've kindly asked for same to this project as well.

@redsk This is the issue reserved to address cross build. You can always ask me to take over the work if you're pretty busy with your stuffs. Thanks in advance!

I use “spark-streaming-kafka-0-10_2.12” to read from kafka.
It seems that it can only used by Structured Streaming.
Can sparkstreaming use this assembly? It doesn't matter if it's not possible, I'll take some time to change my code using Structured Streaming.

Hi ,

As part of manual offset commit we followed your approach and implemented like below .Can you please confirm .

1. As part of onQueryProgress(QueryProgressEvent queryProgress) method we are getting offset and partition details.

2. from queryProgress -> endOffset getting offset and preparing OffsetAndMetadata

3. from queryProgress -> endOffset getting partition and preparing TopicPartition

4. using same group id value to kafkaConsumer which we are processing with spark readStream()

5. finally commiting offset to kafkaConsumer like bleow

        Map<TopicPartition, OffsetAndMetadata> offsets
        kafkaConsumer.commitSync(offsets);
   

We are able to successfully committing mulitple partitions with offsets by above approach.

Next time when spark try to read kafka topic data with readStream() we use startingOffsetDetails() to get startingOffset which are commited through kafka consumer
Dataset dataFrame = spark
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", "127.0.0.1:9092")
.option("subscribe", kafkaTopic)
.option("enable.auto.commit",false)
.option("group.id", groupIdValue)
.option("startingOffsets", startingOffsetDetails())
.option("failOnDataLoss", "false")
.load()
.selectExpr("partition","offset","deserialize(value) as value");

//code to get commited offsets
//the below method returns data like commited offsets {"epeOffsetTpMutltiPartition":{"0":34278,"1":33778}}
private static String startingOffsetDetails(){
String offsetDetails="";
Properties properties = getProperties();
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
List partitions =kafkaConsumer.partitionsFor(kafkaTopic);
StringBuffer sb = new StringBuffer();
sb.append("{"" + kafkaTopic + "":{");
for(PartitionInfo partitionInfo : partitions){
TopicPartition tp = new TopicPartition(kafkaTopic, partitionInfo.partition());
OffsetAndMetadata offsetAndMetadata =kafkaConsumer.committed(tp);
if(null==offsetAndMetadata){
sb.append("""+ partitionInfo.partition() + "":" + 0+"," );
}
else{
sb.append("""+ partitionInfo.partition() + "":" + offsetAndMetadata.offset()+"," );
}
}
sb.deleteCharAt(sb.lastIndexOf(","));
sb.append("}}");
offsetDetails =sb.toString();
System.out.println(offsetDetails);

   return offsetDetails;

}

Can you please confirm the same from your end.

Hi there HeartSaVioR.

You are a legend in this space for providing a library to commit offsets back to Kafka.

I have various businesses that require monitoring of consumers and this is a lifesaver.

Recently I looked into upgrading this library to Spark 3.5.

Through my investigation, I found that they've been significant changes in Spark that now break this library for potential upgrade.

Specifically here: https://github.com/apache/spark/blob/v3.5.0/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceRDD.scala#L32

inputPartition is now inputPartitions which is a Seq of Partitions :(

There are definitely a lot more changes that break the tests as it stands. So will be stuck at Spark 3.0.x.

I've already tried the other versions up to 3.5.x

I'll be reading this code more thoroughly as time allows and seeing if I have the opportunity to make changes and keep this alive.

Otherwise, just letting you know and to also make this visible to you and others.

Hello, I'm a beginner，
I want to ask how to import this program
I didn't find this in Maven

Library not working when spark job deployed in databricks 7.3 LTS with Spark 3.0.1 (Scala -2.12) + (net.heartsavior.spark:spark-sql-kafka-offset-committer_2.12:0.3.0-spark-3.0).

Following warning is displayed in the logs, however same is working in Local spark standalone cluster(with Apache Spark Distribution)

WARN KafkaOffsetCommitterListener: Cannot find query f835de4b-2919-4f60-b849-6082d294e4f7 from active spark session!

Is this library works fine with Kafka SASL SSL?

Please let me know what is the issue.

Or are there plans for supporting 3.1.x?