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

Code before:

      // Listener is updated.
      listener.verify(expectedHighWatermark = 2L)

      // Listener is removed.
      replicaManager.removeListener(tp, listener)

      // Follower fetches up to offset 4.
      fetchPartitionAsFollower(
        replicaManager,
        tidp,
        new FetchRequest.PartitionData(
          Uuid.ZERO_UUID,
          4L,
          0L,
          maxFetchBytes,
          Optional.of(leaderEpoch)
        ),
        replicaId = 1
      )

      // Listener is not updated anymore.
      listener.verify()
    } finally {
      replicaManager.shutdown(checkpointHW = false)
    }
  }

  private def topicsCreateDelta(startId: Int, isStartIdLeader: Boolean, partition:Int = 0, directoryIds: List[Uuid] = List.empty): TopicsDelta = {
    val leader = if (isStartIdLeader) startId else startId + 1
    val delta = new TopicsDelta(TopicsImage.EMPTY)
    delta.replay(new TopicRecord().setName("foo").setTopicId(FOO_UUID))
    val record = partitionRecord(startId, leader, partition)
    if (directoryIds.nonEmpty) {
      record.setDirectories(directoryIds.asJava)
    }
    delta.replay(record)

    delta
  }

  private def partitionRecord(startId: Int, leader: Int, partition: Int = 0) = {
    new PartitionRecord()
      .setPartitionId(partition)
      .setTopicId(FOO_UUID)
      .setReplicas(util.Arrays.asList(startId, startId + 1))
      .setIsr(util.Arrays.asList(startId, startId + 1))
      .setRemovingReplicas(Collections.emptyList())
      .setAddingReplicas(Collections.emptyList())
      .setLeader(leader)
      .setLeaderEpoch(0)

Review comment (refactor):
there are many helpers used to create delta, such as `topicsCreateDelta`, `createFollowerDelta`, `createLeaderDelta`. we might need to clean it down the line

Code after:

      // Listener is removed.
      replicaManager.removeListener(tp, listener)

      // Follower fetches up to offset 4.
      fetchPartitionAsFollower(
        replicaManager,
        tidp,
        new FetchRequest.PartitionData(
          Uuid.ZERO_UUID,
          4L,
          0L,
          maxFetchBytes,
          Optional.of(leaderEpoch)
        ),
        replicaId = 1
      )

      // Listener is not updated anymore.
      listener.verify()
    } finally {
      replicaManager.shutdown(checkpointHW = false)
    }
  }

  private def topicsCreateDelta(startId: Int, isStartIdLeader: Boolean, partitions:List[Int] = List(0), directoryIds: List[Uuid] = List.empty, topicName: String = "foo", topicId: Uuid = FOO_UUID): TopicsDelta = {
    val leader = if (isStartIdLeader) startId else startId + 1
    val delta = new TopicsDelta(TopicsImage.EMPTY)
    delta.replay(new TopicRecord().setName(topicName).setTopicId(topicId))

    partitions.foreach { partition =>
      val record = partitionRecord(startId, leader, partition, topicId)
      if (directoryIds.nonEmpty) {
        record.setDirectories(directoryIds.asJava)
      }
      delta.replay(record)
    }

    delta
  }

  private def partitionRecord(startId: Int, leader: Int, partition: Int = 0, topicId: Uuid = FOO_UUID) = {
    new PartitionRecord()
      .setPartitionId(partition)
      .setTopicId(topicId)
      .setReplicas(util.Arrays.asList(startId, startId + 1))
      .setIsr(util.Arrays.asList(startId, startId + 1))
      .setRemovingReplicas(Collections.emptyList())
      .setAddingReplicas(Collections.emptyList())
      .setLeader(leader)
      .setLeaderEpoch(0)