CosmosNi commented on code in PR #11157:
URL: https://github.com/apache/seatunnel/pull/11157#discussion_r3458437932
##########
seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java:
##########
@@ -195,36 +188,31 @@ private ConsumerMetadata
createConsumerMetadata(ReadonlyConfig readonlyConfig) {
long startOffsetsTimestamp =
readonlyConfig.get(START_MODE_TIMESTAMP);
long currentTimestamp =
System.currentTimeMillis();
- if (startOffsetsTimestamp < 0
- || startOffsetsTimestamp >
currentTimestamp) {
+ // Runtime check: cannot be declarative
(depends on current
+ // time)
+ if (startOffsetsTimestamp >
currentTimestamp) {
throw new IllegalArgumentException(
- "start_mode.timestamp The
value is smaller than 0 or smaller than the current time");
+ "start_mode.timestamp must not
be greater than the current time");
}
consumerMetadata.setStartOffsetsTimestamp(
startOffsetsTimestamp);
if (Objects.nonNull(
readonlyConfig.get(START_MODE_END_TIMESTAMP))) {
long endOffsetsTimestamp =
readonlyConfig.get(START_MODE_END_TIMESTAMP);
- if (endOffsetsTimestamp < 0
- || endOffsetsTimestamp >
currentTimestamp) {
+ // Runtime check: cannot be
declarative (depends on current
+ // time)
+ if (endOffsetsTimestamp >
currentTimestamp) {
throw new IllegalArgumentException(
- "start_mode.endTimestamp
The value is smaller than 0 or smaller than the current time");
+ "start_mode.end_timestamp
must not be greater than the current time");
}
consumerMetadata.setEndOffsetsTimestamp(
endOffsetsTimestamp);
}
break;
case SPECIFIC_OFFSETS:
Review Comment:
This condition is missing.
--
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]
