kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject [kafka] branch trunk updated: MINOR: Do not require request timeout be larger than session timeout (#5246)
Date Mon, 18 Jun 2018 20:43:08 GMT
This is an automated email from the ASF dual-hosted git repository.

jgus pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 347b37e  MINOR: Do not require request timeout be larger than session timeout (#5246)
347b37e is described below

commit 347b37e31945b54fb96fc7dacc2b2deed40e8e5e
Author: Jason Gustafson <jason@confluent.io>
AuthorDate: Mon Jun 18 13:43:01 2018 -0700

    MINOR: Do not require request timeout be larger than session timeout (#5246)
    
    This check was left over from the old consumer logic in which the join group was bound
by the session timeout. Since we use a custom timeout for JoinGroup, this restriction no longer
makes sense.
    
    Reviewers: Ismael Juma <ismael@juma.me.uk>
---
 .../main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java  | 6 ------
 1 file changed, 6 deletions(-)

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 342c559..971b0e7 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
@@ -34,7 +34,6 @@ import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.errors.InterruptException;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.internals.ClusterResourceListeners;
@@ -668,11 +667,6 @@ public class KafkaConsumer<K, V> implements Consumer<K, V>
{
             log.debug("Initializing the Kafka consumer");
             this.requestTimeoutMs = config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG);
             this.defaultApiTimeoutMs = config.getInt(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG);
-            int sessionTimeoutMs = config.getInt(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG);
-            if (this.requestTimeoutMs < sessionTimeoutMs)
-                throw new ConfigException(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG + " ("
+ requestTimeoutMs +
-                                                  ") cannot be lower than  " + ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG
+
-                                                  " (" + sessionTimeoutMs + ")");
             this.time = Time.SYSTEM;
 
             Map<String, String> metricsTags = Collections.singletonMap("client-id",
clientId);


Mime
View raw message