[
https://issues.apache.org/jira/browse/KAFKA-6024?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16396592#comment-16396592
]
ASF GitHub Bot commented on KAFKA-6024:
---------------------------------------
hachikuji closed pull request #4617: KAFKA-6024 Move validation in
KafkaConsumer ahead of acquireAndEnsure…
URL: https://github.com/apache/kafka/pull/4617
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
index 3cd034eff76..81137f3c8dd 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
@@ -966,13 +966,12 @@ public void subscribe(Collection<String> topics) {
*/
@Override
public void subscribe(Pattern pattern, ConsumerRebalanceListener listener)
{
+ if (pattern == null)
+ throw new IllegalArgumentException("Topic pattern to subscribe to
cannot be null");
+
acquireAndEnsureOpen();
try {
- if (pattern == null)
- throw new IllegalArgumentException("Topic pattern to subscribe
to cannot be null");
-
throwIfNoAssignorsConfigured();
-
log.debug("Subscribed to pattern: {}", pattern);
this.subscriptions.subscribe(pattern, listener);
this.metadata.needMetadataForAllTopics(true);
@@ -1337,11 +1336,11 @@ public void commitAsync(final Map<TopicPartition,
OffsetAndMetadata> offsets, Of
*/
@Override
public void seek(TopicPartition partition, long offset) {
+ if (offset < 0)
+ throw new IllegalArgumentException("seek offset must not be a
negative number");
+
acquireAndEnsureOpen();
try {
- if (offset < 0)
- throw new IllegalArgumentException("seek offset must not be a
negative number");
-
log.debug("Seeking to offset {} for partition {}", offset,
partition);
this.subscriptions.seek(partition, offset);
} finally {
@@ -1357,11 +1356,11 @@ public void seek(TopicPartition partition, long offset)
{
* @throws IllegalArgumentException if {@code partitions} is {@code null}
or the provided TopicPartition is not assigned to this consumer
*/
public void seekToBeginning(Collection<TopicPartition> partitions) {
+ if (partitions == null)
+ throw new IllegalArgumentException("Partitions collection cannot
be null");
+
acquireAndEnsureOpen();
try {
- if (partitions == null) {
- throw new IllegalArgumentException("Partitions collection
cannot be null");
- }
Collection<TopicPartition> parts = partitions.size() == 0 ?
this.subscriptions.assignedPartitions() : partitions;
for (TopicPartition tp : parts) {
log.debug("Seeking to beginning of partition {}", tp);
@@ -1383,11 +1382,11 @@ public void seekToBeginning(Collection<TopicPartition>
partitions) {
* @throws IllegalArgumentException if {@code partitions} is {@code null}
or the provided TopicPartition is not assigned to this consumer
*/
public void seekToEnd(Collection<TopicPartition> partitions) {
+ if (partitions == null)
+ throw new IllegalArgumentException("Partitions collection cannot
be null");
+
acquireAndEnsureOpen();
try {
- if (partitions == null) {
- throw new IllegalArgumentException("Partitions collection
cannot be null");
- }
Collection<TopicPartition> parts = partitions.size() == 0 ?
this.subscriptions.assignedPartitions() : partitions;
for (TopicPartition tp : parts) {
log.debug("Seeking to end of partition {}", tp);
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Consider moving validation in KafkaConsumer ahead of call to
> acquireAndEnsureOpen()
> -----------------------------------------------------------------------------------
>
> Key: KAFKA-6024
> URL: https://issues.apache.org/jira/browse/KAFKA-6024
> Project: Kafka
> Issue Type: Improvement
> Reporter: Ted Yu
> Assignee: siva santhalingam
> Priority: Minor
> Fix For: 1.2.0
>
>
> In several methods, parameter validation is done after calling
> acquireAndEnsureOpen() :
> {code}
> public void seek(TopicPartition partition, long offset) {
> acquireAndEnsureOpen();
> try {
> if (offset < 0)
> throw new IllegalArgumentException("seek offset must not be a
> negative number");
> {code}
> Since the value of parameter would not change per invocation, it seems
> performing validation ahead of acquireAndEnsureOpen() call would be better.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
