Repository: apache/kafka
File: core/src/test/scala/unit/kafka/server/ShareGroupHeartbeatRequestTest.scala (Scala)

Code before:
      shareGroupHeartbeatRequest = new ShareGroupHeartbeatRequest.Builder(
        new ShareGroupHeartbeatRequestData()
          .setGroupId("grp")
          .setMemberId(memberId)
          .setMemberEpoch(2)
      ).build()

      // Should receive no error and no assignment changes.
      TestUtils.waitUntilTrue(() => {
        shareGroupHeartbeatResponse = connectAndReceive(shareGroupHeartbeatRequest)
        shareGroupHeartbeatResponse.data.errorCode == Errors.NONE.code
      }, msg = s"Could not get partitions assigned. Last response $shareGroupHeartbeatResponse.")

      // Verify the response. Epoch should not have changed and null assignments determines that no
      // change in old assignment.
      assertEquals(3, shareGroupHeartbeatResponse.data.memberEpoch)
      assertNull(shareGroupHeartbeatResponse.data.assignment)
    } finally {
      admin.close()
    }
  }

  private def connectAndReceive(request: ShareGroupHeartbeatRequest): ShareGroupHeartbeatResponse = {
    val socketServer = cluster.brokerSocketServers().asScala.head
    val listenerName = cluster.clientListener()
    IntegrationTestUtils.connectAndReceive[ShareGroupHeartbeatResponse](request, socketServer.boundPort(listenerName))
  }

  private def increasePartitions[B <: KafkaBroker](admin: Admin,
                                                   topic: String,
                                                   totalPartitionCount: Int
                                                  ): Unit = {
    val newPartitionSet: Map[String, NewPartitions] = Map.apply(topic -> NewPartitions.increaseTo(totalPartitionCount))
    admin.createPartitions(newPartitionSet.asJava)
  }
}


Review comment (suggestion):
```scala
IntegrationTestUtils.connectAndReceive[ShareGroupHeartbeatResponse](request, cluster.boundPorts().get(0))
```

Code after:
      shareGroupHeartbeatRequest = new ShareGroupHeartbeatRequest.Builder(
        new ShareGroupHeartbeatRequestData()
          .setGroupId("grp")
          .setMemberId(memberId)
          .setMemberEpoch(2)
      ).build()

      // Should receive no error and no assignment changes.
      TestUtils.waitUntilTrue(() => {
        shareGroupHeartbeatResponse = connectAndReceive(shareGroupHeartbeatRequest)
        shareGroupHeartbeatResponse.data.errorCode == Errors.NONE.code
      }, msg = s"Could not get partitions assigned. Last response $shareGroupHeartbeatResponse.")

      // Verify the response. Epoch should not have changed and null assignments determines that no
      // change in old assignment.
      assertEquals(3, shareGroupHeartbeatResponse.data.memberEpoch)
      assertNull(shareGroupHeartbeatResponse.data.assignment)
    } finally {
      admin.close()
    }
  }

  private def connectAndReceive(request: ShareGroupHeartbeatRequest): ShareGroupHeartbeatResponse = {
    IntegrationTestUtils.connectAndReceive[ShareGroupHeartbeatResponse](request, cluster.boundPorts().get(0))
  }

  private def increasePartitions[B <: KafkaBroker](admin: Admin,
                                                   topic: String,
                                                   totalPartitionCount: Int
                                                  ): Unit = {
    val newPartitionSet: Map[String, NewPartitions] = Map.apply(topic -> NewPartitions.increaseTo(totalPartitionCount))
    admin.createPartitions(newPartitionSet.asJava)
  }
}