[GitHub] [kafka] feyman2016 commented on a change in pull request #8589: KAFKA-9146: KIP-571 Add option to force delete active members in StreamsResetter

[GitBox]

feyman2016 commented on a change in pull request #8589:
URL: https://github.com/apache/kafka/pull/8589#discussion_r429082038



##########
File path: 
streams/src/test/java/org/apache/kafka/streams/integration/AbstractResetIntegrationTest.java
##########
@@ -261,6 +261,43 @@ public void 
shouldNotAllowToResetWhenIntermediateTopicAbsent() throws Exception
         Assert.assertEquals(1, exitCode);
     }
 
+    public void testResetWhenLongSessionTimeoutConfiguredWithForceOption() 
throws Exception {
+        appID = testId + "-with-force-option";
+        streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
+        streamsConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "" + 
STREAMS_CONSUMER_TIMEOUT * 100);

Review comment:
       Without the `"" +` to convert the value to String, we will get exception 
like: it is because `STREAMS_CONSUMER_TIMEOUT = 2000L`, `""+` is widely used in 
this test, just follow it here without any change to not enlarge the scope of 
this PR, I can help to create a Jira to enhance it if we think this workaround 
is not quite intuitive~
   
   ```
   org.apache.kafka.common.config.ConfigException: Invalid value 200000 for 
configuration session.timeout.ms: Expected value to be a 32-bit integer, but it 
was a java.lang.Long
           at 
org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:672)
           at 
org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:474)
           at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:467)
           at 
org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:108)
           at 
org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:129)
           at 
org.apache.kafka.clients.consumer.ConsumerConfig.<init>(ConsumerConfig.java:606)
           at 
org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:630)
           at 
org.apache.kafka.streams.processor.internals.DefaultKafkaClientSupplier.getRestoreConsumer(DefaultKafkaClientSupplier.java:56)
           at 
org.apache.kafka.streams.processor.internals.StreamThread.create(StreamThread.java:313)
           at 
org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:766)
           at 
org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:652)
           at 
org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:562)
           at 
org.apache.kafka.streams.integration.AbstractResetIntegrationTest.testResetWhenLongSessionTimeoutConfiguredWithForceOption(AbstractResetIntegrationTest.java:270)
           at 
org.apache.kafka.streams.integration.ResetIntegrationTest.testResetWhenLongSessionTimeoutConfiguredWithForceOption(ResetIntegrationTest.java:77)
   
   ``` 




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org