navina commented on a change in pull request #8309:
URL: https://github.com/apache/pinot/pull/8309#discussion_r821266312
##########
File path:
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConnectionHandler.java
##########
@@ -56,6 +57,7 @@ public KafkaPartitionLevelConnectionHandler(String clientId,
StreamConfig stream
consumerProp.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
_config.getBootstrapHosts());
consumerProp.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
consumerProp.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
BytesDeserializer.class.getName());
+ consumerProp.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
OffsetResetStrategy.EARLIEST.toString().toLowerCase());
Review comment:
Can you make this user-configurable ? we don't know what caused the OOR
and what behavior how the user expects. Also, since this changes the default
behavior, it is not a backward compatible change.
I would recommend making the default as "LATEST" and override, if the user
has provided an override.
##########
File path:
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/utils/MiniKafkaCluster.java
##########
@@ -112,4 +115,9 @@ public void deleteTopic(String topicName)
throws ExecutionException, InterruptedException {
_adminClient.deleteTopics(Collections.singletonList(topicName)).all().get();
}
+
+ public void deleteRecordsBeforeOffset(String topicName, int partitionId,
long offset) {
+ _adminClient.deleteRecords(
Review comment:
oo.. have never used this API. I am curious what happens when the offset
provided here is invalid - does it throw or just ignores the call.
--
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: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
