[KAFKA-16659] KafkaConsumer#position() does not respect wakup when group protocol is CONSUMER

• Type: Bug
• Status: Resolved
• Priority: Minor
• Resolution: Fixed
• Fix Version/s: 3.8.0
• Assignee: PoAn Yang
• Reporter: Chia-Ping Tsai
• Created: 02/May/24 21:40
• Updated: 06/May/24 00:45
• Resolved: 06/May/24 00:45

Description

see following test

  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
  @Timeout(15)
  def testPositionRespectsWakeup(quorum: String, groupProtocol: String): Unit = {
    val topicPartition = new TopicPartition(topic, 15)
    val consumer = createConsumer()
    consumer.assign(List(topicPartition).asJava)

    CompletableFuture.runAsync { () =>
      TimeUnit.SECONDS.sleep(1)
      consumer.wakeup()
    }

    assertThrows(classOf[WakeupException], () => consumer.position(topicPartition, Duration.ofSeconds(3)))
  }

Issue Links

• links to: GitHub Pull Request #15853