mjsax commented on code in PR #20347:
URL: https://github.com/apache/kafka/pull/20347#discussion_r2274987340
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java:
##########
@@ -814,29 +838,56 @@ private void createStateForRestoration(final String
changelogTopic, final int st
}
}
- private void setCommittedOffset(final String topic, final int limitDelta) {
- final Properties consumerConfig = new Properties();
- consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers());
- consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, appId);
- consumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, "commit-consumer");
- consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
IntegerDeserializer.class);
- consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
IntegerDeserializer.class);
-
- final Consumer<Integer, Integer> consumer = new
KafkaConsumer<>(consumerConfig);
- final List<TopicPartition> partitions = asList(
- new TopicPartition(topic, 0),
- new TopicPartition(topic, 1));
-
- consumer.assign(partitions);
- consumer.seekToEnd(partitions);
-
- for (final TopicPartition partition : partitions) {
- final long position = consumer.position(partition);
- consumer.seek(partition, position - limitDelta);
+ private void setCommittedOffset(final String topic, final int limitDelta,
final boolean useNewProtocol) {
+ if (useNewProtocol) {
+ final Properties consumerConfig = new Properties();
+ consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers());
+ consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, appId);
+ consumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG,
"commit-consumer");
+ consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
IntegerDeserializer.class);
+ consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
IntegerDeserializer.class);
+
+ try (final Consumer<Integer, Integer> consumer = new
KafkaConsumer<>(consumerConfig)) {
+ final List<TopicPartition> partitions = asList(
+ new TopicPartition(topic, 0),
+ new TopicPartition(topic, 1));
+
+ consumer.assign(partitions);
+ consumer.seekToEnd(partitions);
+
+ for (final TopicPartition partition : partitions) {
+ final long position = consumer.position(partition);
+ consumer.seek(partition, position - limitDelta);
+ }
+
+ consumer.commitSync();
+ }
}
Review Comment:
Should there be an `else` here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
