File tree 2 files changed +4
-3
lines changed
src/main/java/com/baeldung/kafka/exactlyonce
2 files changed +4
-3
lines changed Original file line number Diff line number Diff line change 124
124
125
125
<properties >
126
126
<jackson .version>2.15.4</jackson .version> <!-- Upgrading to version >= 2.16 breaks tests -->
127
- <kafka .version>3.4 .0</kafka .version>
127
+ <kafka .version>3.9 .0</kafka .version>
128
128
<testcontainers-kafka .version>1.19.3</testcontainers-kafka .version>
129
129
<testcontainers-jupiter .version>1.15.3</testcontainers-jupiter .version>
130
130
<org .apache.spark.spark-core.version>2.4.8</org .apache.spark.spark-core.version>
131
131
<com .datastax.spark.spark-cassandra-connector.version>2.5.2</com .datastax.spark.spark-cassandra-connector.version>
132
132
<com .datastax.spark.spark-cassandra-connector-java.version>1.6.0-M1</com .datastax.spark.spark-cassandra-connector-java.version>
133
133
</properties >
134
134
135
- </project >
135
+ </project >
Original file line number Diff line number Diff line change 2
2
3
3
import org .apache .kafka .clients .consumer .ConsumerRecord ;
4
4
import org .apache .kafka .clients .consumer .ConsumerRecords ;
5
+ import org .apache .kafka .clients .consumer .ConsumerGroupMetadata ;
5
6
import org .apache .kafka .clients .consumer .KafkaConsumer ;
6
7
import org .apache .kafka .clients .consumer .OffsetAndMetadata ;
7
8
import org .apache .kafka .clients .producer .KafkaProducer ;
@@ -63,7 +64,7 @@ public static void main(String[] args) {
63
64
offsetsToCommit .put (partition , new OffsetAndMetadata (offset + 1 ));
64
65
}
65
66
66
- producer .sendOffsetsToTransaction (offsetsToCommit , CONSUMER_GROUP_ID );
67
+ producer .sendOffsetsToTransaction (offsetsToCommit , new ConsumerGroupMetadata ( CONSUMER_GROUP_ID ) );
67
68
producer .commitTransaction ();
68
69
69
70
}
You can’t perform that action at this time.
0 commit comments