You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lianetm (via GitHub)" <gi...@apache.org> on 2023/10/05 15:55:54 UTC
[PR] MINOR - KAFKA-15550: Validation for negative target times in offsetsForTimes [kafka]
lianetm opened a new pull request, #14503:
URL: https://github.com/apache/kafka/pull/14503
The current KafkaConsumer `offsetsForTimes` fails with `IllegalArgumentException` if negative target timestamps are provided as arguments. This PR includes the same validation and tests for the new consumer implementation.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR - KAFKA-15550: Validation for negative target times in offsetsForTimes [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #14503:
URL: https://github.com/apache/kafka/pull/14503#discussion_r1357057172
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java:
##########
@@ -322,6 +323,25 @@ public void testOffsetsForTimesOnNullPartitions() {
Duration.ofMillis(1)));
}
+ @Test
+ public void testOffsetsForTimesFailsOnNegativeTargetTimes() {
+ PrototypeAsyncConsumer<?, ?> consumer = newConsumer(time, new StringDeserializer(), new StringDeserializer());
+ assertThrows(IllegalArgumentException.class,
+ () -> consumer.offsetsForTimes(Collections.singletonMap(new TopicPartition(
+ "topic1", 1), ListOffsetsRequest.EARLIEST_TIMESTAMP),
+ Duration.ofMillis(1)));
+
+ assertThrows(IllegalArgumentException.class,
+ () -> consumer.offsetsForTimes(Collections.singletonMap(new TopicPartition(
Review Comment:
Consistency with the old consumer, say no more!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR - KAFKA-15550: Validation for negative target times in offsetsForTimes [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #14503:
URL: https://github.com/apache/kafka/pull/14503
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR - KAFKA-15550: Validation for negative target times in offsetsForTimes [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14503:
URL: https://github.com/apache/kafka/pull/14503#discussion_r1356866475
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java:
##########
@@ -322,6 +323,25 @@ public void testOffsetsForTimesOnNullPartitions() {
Duration.ofMillis(1)));
}
+ @Test
+ public void testOffsetsForTimesFailsOnNegativeTargetTimes() {
+ PrototypeAsyncConsumer<?, ?> consumer = newConsumer(time, new StringDeserializer(), new StringDeserializer());
+ assertThrows(IllegalArgumentException.class,
+ () -> consumer.offsetsForTimes(Collections.singletonMap(new TopicPartition(
+ "topic1", 1), ListOffsetsRequest.EARLIEST_TIMESTAMP),
+ Duration.ofMillis(1)));
+
+ assertThrows(IllegalArgumentException.class,
+ () -> consumer.offsetsForTimes(Collections.singletonMap(new TopicPartition(
Review Comment:
These constants are defined public in the [ListOffsetsRequest](https://github.com/apache/kafka/blob/4cf86c5d2f6355c5684ecedeb1bf96e90877244d/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetsRequest.java#L41-L43).
You're right that the validation is indeed for all negative values, but this test is what I expect could be the most common misuse of this API (a user trying to retrieve offsets using an "abstract" timestamp like EARLIEST or LATEST). This also was the initial motivation for the validation in the old consumer with [this](https://github.com/apache/kafka/blob/4cf86c5d2f6355c5684ecedeb1bf96e90877244d/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java#L2115) comment, so aligning with it. Makes sense?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR - KAFKA-15550: Validation for negative target times in offsetsForTimes [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #14503:
URL: https://github.com/apache/kafka/pull/14503#discussion_r1356610621
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -252,8 +255,10 @@ private boolean updateFetchPositionsIfNeeded(final Timer timer) {
// are partitions with a missing position, then we will raise a NoOffsetForPartitionException exception.
subscriptions.resetInitializingPositions();
- // Finally send an asynchronous request to look up and update the positions of any
- // partitions which are awaiting reset.
+ // Reset positions using partitions offsets retrieved from the leader, for any partitions
Review Comment:
```suggestion
// Reset positions using partition offsets retrieved from the leader, for any partitions
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -235,15 +235,18 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
* defined
*/
private boolean updateFetchPositionsIfNeeded(final Timer timer) {
- // If any partitions have been truncated due to a leader change, we need to validate the offsets
+ // Validate positions using the partition leader end offsets, to detect if any partition
+ // has been truncated due to a leader change. This will trigger an OffsetForLeaderEpoch
+ // request, retrieve the partition end offsets, and validate the current position against it.
ValidatePositionsApplicationEvent validatePositionsEvent = new ValidatePositionsApplicationEvent();
eventHandler.add(validatePositionsEvent);
- // If there are any partitions which do not have a valid position and are not
- // awaiting reset, then we need to fetch committed offsets. We will only do a
- // coordinator lookup if there are partitions which have missing positions, so
- // a consumer with manually assigned partitions can avoid a coordinator dependence
- // by always ensuring that assigned partitions have an initial position.
+ // Reset positions using committed offsets retrieved from the group coordinator, for any
+ // partitions which do not have a valid position and are not awaiting reset. We
+ // will only do a coordinator lookup if there are partitions which have missing
+ // positions, so a consumer with manually assigned partitions can avoid a coordinator
+ // dependence by always ensuring that assigned partitions have an initial position. This
+ // will trigger an OffsetFetch request and update positions with the offsets retrieved.
Review Comment:
Could you swap sentence 2 and 3 in this paragraph? It sounds like having manual partition assignment with an initial position will trigger an `OffsetFetch` request in this order.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -235,15 +235,18 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
* defined
*/
private boolean updateFetchPositionsIfNeeded(final Timer timer) {
- // If any partitions have been truncated due to a leader change, we need to validate the offsets
+ // Validate positions using the partition leader end offsets, to detect if any partition
+ // has been truncated due to a leader change. This will trigger an OffsetForLeaderEpoch
+ // request, retrieve the partition end offsets, and validate the current position against it.
ValidatePositionsApplicationEvent validatePositionsEvent = new ValidatePositionsApplicationEvent();
eventHandler.add(validatePositionsEvent);
- // If there are any partitions which do not have a valid position and are not
- // awaiting reset, then we need to fetch committed offsets. We will only do a
- // coordinator lookup if there are partitions which have missing positions, so
- // a consumer with manually assigned partitions can avoid a coordinator dependence
- // by always ensuring that assigned partitions have an initial position.
+ // Reset positions using committed offsets retrieved from the group coordinator, for any
+ // partitions which do not have a valid position and are not awaiting reset. We
Review Comment:
```suggestion
// partitions which do not have a valid position and are not awaiting reset. We
```
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java:
##########
@@ -322,6 +323,25 @@ public void testOffsetsForTimesOnNullPartitions() {
Duration.ofMillis(1)));
}
+ @Test
+ public void testOffsetsForTimesFailsOnNegativeTargetTimes() {
+ PrototypeAsyncConsumer<?, ?> consumer = newConsumer(time, new StringDeserializer(), new StringDeserializer());
+ assertThrows(IllegalArgumentException.class,
+ () -> consumer.offsetsForTimes(Collections.singletonMap(new TopicPartition(
+ "topic1", 1), ListOffsetsRequest.EARLIEST_TIMESTAMP),
+ Duration.ofMillis(1)));
+
+ assertThrows(IllegalArgumentException.class,
+ () -> consumer.offsetsForTimes(Collections.singletonMap(new TopicPartition(
Review Comment:
Why do we test this for these three constants and not for `EARLIEST_LOCAL_TIMESTAMP`?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR - KAFKA-15550: Validation for negative target times in offsetsForTimes [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #14503:
URL: https://github.com/apache/kafka/pull/14503#discussion_r1356663540
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java:
##########
@@ -322,6 +323,25 @@ public void testOffsetsForTimesOnNullPartitions() {
Duration.ofMillis(1)));
}
+ @Test
+ public void testOffsetsForTimesFailsOnNegativeTargetTimes() {
+ PrototypeAsyncConsumer<?, ?> consumer = newConsumer(time, new StringDeserializer(), new StringDeserializer());
+ assertThrows(IllegalArgumentException.class,
+ () -> consumer.offsetsForTimes(Collections.singletonMap(new TopicPartition(
+ "topic1", 1), ListOffsetsRequest.EARLIEST_TIMESTAMP),
+ Duration.ofMillis(1)));
+
+ assertThrows(IllegalArgumentException.class,
+ () -> consumer.offsetsForTimes(Collections.singletonMap(new TopicPartition(
Review Comment:
Actually, under which circumstances would a user even use these constants? All uses seem to be internal, if I see correctly. Checking non-negative makes sense, just not sure why we are checking for these (internal?) constants specifically.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR - KAFKA-15550: Validation for negative target times in offsetsForTimes [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #14503:
URL: https://github.com/apache/kafka/pull/14503#issuecomment-1761075787
Test failures unrelated
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR - KAFKA-15550: Validation for negative target times in offsetsForTimes [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14503:
URL: https://github.com/apache/kafka/pull/14503#discussion_r1356848021
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -235,15 +235,18 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
* defined
*/
private boolean updateFetchPositionsIfNeeded(final Timer timer) {
- // If any partitions have been truncated due to a leader change, we need to validate the offsets
+ // Validate positions using the partition leader end offsets, to detect if any partition
+ // has been truncated due to a leader change. This will trigger an OffsetForLeaderEpoch
+ // request, retrieve the partition end offsets, and validate the current position against it.
ValidatePositionsApplicationEvent validatePositionsEvent = new ValidatePositionsApplicationEvent();
eventHandler.add(validatePositionsEvent);
- // If there are any partitions which do not have a valid position and are not
- // awaiting reset, then we need to fetch committed offsets. We will only do a
- // coordinator lookup if there are partitions which have missing positions, so
- // a consumer with manually assigned partitions can avoid a coordinator dependence
- // by always ensuring that assigned partitions have an initial position.
+ // Reset positions using committed offsets retrieved from the group coordinator, for any
+ // partitions which do not have a valid position and are not awaiting reset. We
+ // will only do a coordinator lookup if there are partitions which have missing
+ // positions, so a consumer with manually assigned partitions can avoid a coordinator
+ // dependence by always ensuring that assigned partitions have an initial position. This
+ // will trigger an OffsetFetch request and update positions with the offsets retrieved.
Review Comment:
Agree. Done.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR - KAFKA-15550: Validation for negative target times in offsetsForTimes [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #14503:
URL: https://github.com/apache/kafka/pull/14503#issuecomment-1756011144
hey @cadonna, @lucasbru. This is a very small PR adding a missing API validation to the new consumer `offsetsForTimes` functionality.
It also includes some improved comments for the `updateFetchPositions`. It could all be useful to take a look at the new code if you have some time. Thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR - KAFKA-15550: Validation for negative target times in offsetsForTimes [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm closed pull request #14503: MINOR - KAFKA-15550: Validation for negative target times in offsetsForTimes
URL: https://github.com/apache/kafka/pull/14503
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR - KAFKA-15550: Validation for negative target times in offsetsForTimes [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #14503:
URL: https://github.com/apache/kafka/pull/14503#issuecomment-1759658605
hey @lucasbru, thanks a lot for all the comments, all addressed.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org