[GitHub] [pinot] navina commented on a change in pull request #8309: KafkaConsumer to reset offset to earliest when fetch position out of range

[GitBox <gi...@apache.org>]

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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org