richardstartin commented on a change in pull request #8057:
URL: https://github.com/apache/pinot/pull/8057#discussion_r790593249
##########
File path:
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java
##########
@@ -63,6 +69,28 @@ public StreamPartitionMsgOffset
fetchStreamPartitionOffset(OffsetCriteria offset
offset =
_consumer.beginningOffsets(Collections.singletonList(_topicPartition),
Duration.ofMillis(timeoutMillis))
.get(_topicPartition);
+ } else if (offsetCriteria.isPeriod()) {
+ String offsetString = offsetCriteria.getOffsetString();
+ Long periodToMillis = TimeUtils.convertPeriodToMillis(offsetString);
+ TopicPartition tp = new TopicPartition(_topic, _partition);
+ _offsetsForTimes.put(tp, System.currentTimeMillis() - periodToMillis);
Review comment:
Please use an injected `java.time.Clock` so the time can be controlled
for testing purposes. Injecting `Clock.systemClock()` and calling
`Clock.millis()` is equivalent to this logic, but allows you to inject a fixed
clock for testing purposes.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
