KAFKA-19807: Add RPC-level integration tests for StreamsGroupHeartbeat [1/2] #20757

lucliu1108 · 2025-10-23T04:31:26Z

What

Ticket: https://issues.apache.org/jira/browse/KAFKA-19807

Add integration test similar to ShareGroupHeartbeatRequestTest and
ConsumerGroupHeartbeatRequestTest for StreamsGroupHeartbeat

Reviewers: @lucasbru

Copilot

Pull Request Overview

This PR adds RPC-level integration tests for the StreamsGroupHeartbeat API, following the same testing pattern as ShareGroupHeartbeatRequestTest and ConsumerGroupHeartbeatRequestTest.

Key changes:

Implements comprehensive integration tests for the StreamsGroupHeartbeat RPC API
Tests various scenarios including API version validation, feature flag checks, multi-member coordination, and error handling
Validates proper behavior when topics are missing and subsequently created

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

Copilot · 2025-10-23T07:48:30Z

core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala

+      new ClusterFeature(feature = Feature.STREAMS_VERSION, version = 0)
+    )
+  )
+  def testStreamsGroupHeartbeatIsInaccessableWhenDisabledByFeatureConfig(): Unit = {


Corrected spelling of 'Inaccessable' to 'Inaccessible' in test method name.

Suggested change

def testStreamsGroupHeartbeatIsInaccessableWhenDisabledByFeatureConfig(): Unit = {

def testStreamsGroupHeartbeatIsInaccessibleWhenDisabledByFeatureConfig(): Unit = {

Copilot · 2025-10-23T07:48:30Z

core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala

+      new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic,consumer"),
+    )
+  )
+  def testStreamsGroupHeartbeatIsInaccessableWhenDisabledByStaticGroupCoordinatorProtocolConfig(): Unit = {


Corrected spelling of 'Inaccessable' to 'Inaccessible' in test method name.

Suggested change

def testStreamsGroupHeartbeatIsInaccessableWhenDisabledByStaticGroupCoordinatorProtocolConfig(): Unit = {

def testStreamsGroupHeartbeatIsInaccessibleWhenDisabledByStaticGroupCoordinatorProtocolConfig(): Unit = {

Copilot · 2025-10-23T07:48:30Z

core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala

+  }
+
+  @ClusterTest
+  def tesStreamsGroupHeartbeatIsAccessibleWhenNewGroupCoordinatorIsEnabledTopicNotExistFirst(): Unit = {


Corrected spelling of 'tesStreamsGroupHeartbeat' to 'testStreamsGroupHeartbeat' in test method name (missing 't').

Suggested change

def tesStreamsGroupHeartbeatIsAccessibleWhenNewGroupCoordinatorIsEnabledTopicNotExistFirst(): Unit = {

def testStreamsGroupHeartbeatIsAccessibleWhenNewGroupCoordinatorIsEnabledTopicNotExistFirst(): Unit = {

Copilot · 2025-10-23T07:48:30Z

core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala

+  }
+
+  @ClusterTest
+  def tesStreamsGroupHeartbeatForMultipleMembers(): Unit = {


Corrected spelling of 'tesStreamsGroupHeartbeat' to 'testStreamsGroupHeartbeat' in test method name (missing 't').

Suggested change

def tesStreamsGroupHeartbeatForMultipleMembers(): Unit = {

def testStreamsGroupHeartbeatForMultipleMembers(): Unit = {

lucasbru

Thanks for the PR! Looking very good already. Could we add a couple more tests:

As in ConsumerGroupHeartbeatRequestTest, we could set a dynamic group config. In particular, I think it would be good to test setting streams.num.standby.replicas. We can set the config from 0 to 1 and check if we get a standby task in the assignment after heartbeating again. Note that we need 2 members with two different process IDs for this to work. This would also test your earlier change that triggers a rebalance upon config change.
It would be good to have a basic test to check that internal topics are created. So essentially heartbeat with a topology that contains internal topics (changelog topic, repartition topic). Then wait that the internal topics eventually appear.
Having equivalents for ShareGroupHeartbeatRequestTest.testMemberJoiningAndExpiring and ShareGroupHeartbeatRequestTest.testGroupCoordinatorChange would be very nice to have.

lucasbru · 2025-10-23T12:45:37Z

core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala

+      }, "StreamsGroupHeartbeatRequest did not succeed within the timeout period.")
+
+      // Verify the response
+      assert(streamsGroupHeartbeatResponse != null, "StreamsGroupHeartbeatResponse should not be null")


This is a streams assert. Maybe we should use junit asserts only to be consistent. Here we'd use assertNotNull.

lucliu1108 · 2025-10-23T16:04:45Z

Thanks @lucasbru for the review!

I will add the other tests in a follow-up PR.

lucasbru

Sorry, I have two more thing I noticed while reviewing the second PR

lucasbru · 2025-10-26T12:23:53Z

core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala

+    // Test with streams.version = 0, the API is disabled at server level
+    val topology = new StreamsGroupHeartbeatRequestData.Topology()
+      .setEpoch(1)
+      .setSubtopologies(java.util.Collections.emptyList())


java.util.Collections is already imported, so you do not need to add the FQN here.

Also in many other places.

You could also consider List.of() instead which is more readable.

lucasbru · 2025-10-26T12:26:30Z

core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala

+        .setStandbyTasks(java.util.Collections.emptyList())
+        .setWarmupTasks(java.util.Collections.emptyList())
+        .setTopology(topology)
+    ).build(0)


We should probably always build the latest version?

).build(ApiKeys.STREAMS_GROUP_HEARTBEAT.latestVersion(isUnstableApiEnabled))

lucliu1108 added 5 commits October 20, 2025 13:42

add basic tests

b5cb287

revise test assert

254d77a

Merge branch 'trunk' into KAFKA-19807

355e15f

add more tests

3567319

revise tests

b94657a

github-actions bot added triage PRs from the community core Kafka Broker tests Test fixes (including flaky tests) labels Oct 23, 2025

lucasbru requested review from Copilot and lucasbru October 23, 2025 07:47

lucasbru self-assigned this Oct 23, 2025

Copilot AI reviewed Oct 23, 2025

View reviewed changes

lucasbru reviewed Oct 23, 2025

View reviewed changes

address comments

9e6e277

lucliu1108 changed the title ~~KAFKA-19807: Add RPC-level integration tests for StreamsGroupHeartbeat~~ KAFKA-19807: Add RPC-level integration tests for StreamsGroupHeartbeat [1/2] Oct 23, 2025

lucliu1108 requested a review from lucasbru October 23, 2025 15:27

github-actions bot removed the triage PRs from the community label Oct 24, 2025

lucliu1108 mentioned this pull request Oct 24, 2025

KAFKA-19807: Add RPC-level integration tests for StreamsGroupHeartbeat [2/2] #20762

Open

add file header

129912a

lucasbru reviewed Oct 26, 2025

View reviewed changes

lucliu1108 added 2 commits October 26, 2025 17:11

use streamsGroupHeartbeat helper method

ced277d

remove default param definition for all methods

4f758f2

lucliu1108 requested a review from lucasbru October 26, 2025 22:34

revise java list import

3f0e0ee

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Uh oh!

KAFKA-19807: Add RPC-level integration tests for StreamsGroupHeartbeat [1/2] #20757

KAFKA-19807: Add RPC-level integration tests for StreamsGroupHeartbeat [1/2] #20757

lucliu1108 commented Oct 23, 2025 •

edited by github-actions bot

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Oct 23, 2025

Uh oh!

Copilot AI Oct 23, 2025

Uh oh!

Copilot AI Oct 23, 2025

Uh oh!

Copilot AI Oct 23, 2025

Uh oh!

lucasbru left a comment

Uh oh!

lucasbru Oct 23, 2025

Uh oh!

lucliu1108 commented Oct 23, 2025

Uh oh!

lucasbru left a comment

Uh oh!

lucasbru Oct 26, 2025

Uh oh!

lucasbru Oct 26, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

	def testStreamsGroupHeartbeatIsInaccessableWhenDisabledByFeatureConfig(): Unit = {
	def testStreamsGroupHeartbeatIsInaccessibleWhenDisabledByFeatureConfig(): Unit = {

	def tesStreamsGroupHeartbeatIsAccessibleWhenNewGroupCoordinatorIsEnabledTopicNotExistFirst(): Unit = {
	def testStreamsGroupHeartbeatIsAccessibleWhenNewGroupCoordinatorIsEnabledTopicNotExistFirst(): Unit = {

	def tesStreamsGroupHeartbeatForMultipleMembers(): Unit = {
	def testStreamsGroupHeartbeatForMultipleMembers(): Unit = {

Uh oh!

KAFKA-19807: Add RPC-level integration tests for StreamsGroupHeartbeat [1/2] #20757

Are you sure you want to change the base?

KAFKA-19807: Add RPC-level integration tests for StreamsGroupHeartbeat [1/2] #20757

Conversation

lucliu1108 commented Oct 23, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Uh oh!

Copilot AI Oct 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 23, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

lucasbru Oct 23, 2025

Choose a reason for hiding this comment

Uh oh!

lucliu1108 commented Oct 23, 2025

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

lucasbru Oct 26, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru Oct 26, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

lucliu1108 commented Oct 23, 2025 •

edited by github-actions bot

Loading