[jira] [Created] (KAFKA-17581) AsyncKafkaConsumer can't unsubscribe invalid topics

["PoAn Yang (Jira)" <ji...@apache.org>]

PoAn Yang created KAFKA-17581:
---------------------------------

             Summary: AsyncKafkaConsumer can't unsubscribe invalid topics
                 Key: KAFKA-17581
                 URL: https://issues.apache.org/jira/browse/KAFKA-17581
             Project: Kafka
          Issue Type: Bug
            Reporter: PoAn Yang
            Assignee: PoAn Yang


When consumer subscribes an invalid topic name like " this is test", classic consumer can unsubscribe without error. However, async consumer can't. We can use following integration test to validate:

 
{code:java}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
@MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
def testSubscribeInvalidTopic(quorum: String, groupProtocol: String): Unit = {
  // Invalid topic name due to space
  val invalidTopicName = "topic abc"
  val consumer = createConsumer()

  consumer.subscribe(List(invalidTopicName).asJava)

  var exception : InvalidTopicException = null
  TestUtils.waitUntilTrue(() => {
    try consumer.poll(Duration.ofMillis(500)) catch {
      case e : InvalidTopicException => exception = e
      case e : Throwable => fail(s"An InvalidTopicException should be thrown. But ${e.getClass} is thrown")
    }
    exception != null
  }, waitTimeMs = 5000, msg = "An InvalidTopicException should be thrown.")

  assertEquals(s"Invalid topics: [${invalidTopicName}]", exception.getMessage)
  // AsyncKafkaConsumer sends request in background thread. Wait enough time to send next request.
  Thread.sleep(1000)
  assertDoesNotThrow(new Executable {
    override def execute(): Unit = consumer.unsubscribe()
  })
}{code}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)