Jackie-Jiang commented on code in PR #9193:
URL: https://github.com/apache/pinot/pull/9193#discussion_r947313065
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/OffsetCriteria.java:
##########
@@ -172,17 +196,21 @@ public OffsetCriteria withOffsetString(String
offsetString) {
_offsetCriteria.setOffsetType(OffsetType.SMALLEST);
} else if (offsetString.equalsIgnoreCase(OffsetType.LARGEST.toString()))
{
_offsetCriteria.setOffsetType(OffsetType.LARGEST);
- } else {
+ } else if (TimeUtils.isPeriodValid(offsetString)) {
try {
- Long periodToMillis = TimeUtils.convertPeriodToMillis(offsetString);
+ long periodToMillis = TimeUtils.convertPeriodToMillis(offsetString);
if (periodToMillis >= 0) {
_offsetCriteria.setOffsetType(OffsetType.PERIOD);
} else {
_offsetCriteria.setOffsetType(OffsetType.CUSTOM);
}
- } catch (Exception e) {
- _offsetCriteria.setOffsetType(OffsetType.CUSTOM);
+ } catch (Exception ignore) {
Review Comment:
Since we already check if the period is valid, we should just remove the
try-catch
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java:
##########
@@ -63,6 +70,28 @@ public StreamPartitionMsgOffset
fetchStreamPartitionOffset(OffsetCriteria offset
offset =
_consumer.beginningOffsets(Collections.singletonList(_topicPartition),
Duration.ofMillis(timeoutMillis))
.get(_topicPartition);
+ } else if (offsetCriteria.isPeriod()) {
+ OffsetAndTimestamp offsetAndTimestamp =
_consumer.offsetsForTimes(Collections.singletonMap(_topicPartition,
+ Clock.systemUTC().millis() -
TimeUtils.convertPeriodToMillis(offsetCriteria.getOffsetString())))
+ .get(_topicPartition);
+ if (offsetAndTimestamp == null) {
+ LOGGER.warn("initial offset type is period and its value evaluates
to null hence proceeding with endOffset");
Review Comment:
Suggest also logging the topic partition and offset string to provide more
info for debugging
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java:
##########
@@ -63,6 +70,28 @@ public StreamPartitionMsgOffset
fetchStreamPartitionOffset(OffsetCriteria offset
offset =
_consumer.beginningOffsets(Collections.singletonList(_topicPartition),
Duration.ofMillis(timeoutMillis))
.get(_topicPartition);
+ } else if (offsetCriteria.isPeriod()) {
+ OffsetAndTimestamp offsetAndTimestamp =
_consumer.offsetsForTimes(Collections.singletonMap(_topicPartition,
+ Clock.systemUTC().millis() -
TimeUtils.convertPeriodToMillis(offsetCriteria.getOffsetString())))
+ .get(_topicPartition);
+ if (offsetAndTimestamp == null) {
+ LOGGER.warn("initial offset type is period and its value evaluates
to null hence proceeding with endOffset");
+ offset =
_consumer.endOffsets(Collections.singletonList(_topicPartition),
Duration.ofMillis(timeoutMillis))
+ .get(_topicPartition);
+ } else {
+ offset = offsetAndTimestamp.offset();
+ }
+ } else if (offsetCriteria.isTimestamp()) {
+ OffsetAndTimestamp offsetAndTimestamp =
_consumer.offsetsForTimes(Collections.singletonMap(_topicPartition,
+
TimeUtils.convertTimestampToMillis(offsetCriteria.getOffsetString()))).get(_topicPartition);
+ if (offsetAndTimestamp == null) {
+ LOGGER.warn("initial offset type is timestamp and its value
evaluates "
Review Comment:
Same 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: 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
