// REPRODUCES the "sub-seal gap" silent-data-loss bug.
  //
  // Setup of a consolidating partition:
  //   - classic-to-diskless seal  = 1000  (offsets [1000, ...) live in object storage / diskless)
  //   - local classic prefix      = [0, 1000) and spans multiple local log segments
  //   - local logEndOffset (LEO)  = 1000
  //   - localLogStartOffset       = 0     (nothing tiered, so the tiered-range guard is a no-op)
  //
  // A consumer fetches at offset 480, which lands in an *earlier* local segment. LocalLog.read
  // serves from a single segment, so the local read returns only [480, 500) (a segment boundary
  // well below the seal). Because that is smaller than minBytes, the consolidating path triggers
  // a diskless supplement, computed by buildConsolidationSupplementFetchInfos as
  // lastBatch().nextOffset() == 500.
  //
  // But 500 is BELOW the seal (1000): the diskless control plane holds no batches for offsets
  // below the seal, so a FindBatch at 500 returns the first available batch — at the seal, 1000.
  // (The fixed FetchHandler stub returns records starting at 1000 regardless of the requested
  // offset, faithfully modelling that tailSet/"first batch at-or-after" control-plane behavior.)
  //
  // The merge therefore stitches local [480, 500) directly onto diskless [1000, ...), and the
  // committed range [500, 1000) is silently dropped from the consumer-visible response.
  @Test
  def testFetchConsolidatingSupplementBelowSealSilentlySkipsClassicPrefix(): Unit = {
    val seal = 1000L
    val localLeo = 1000L

    // Local single-segment read: offsets [480, 500). 20 records in one batch -> nextOffset == 500.
    val localRecordList = (0 until 20).map(i => new SimpleRecord(i.toLong, s"local-$i".getBytes())).toArray
    val localMemRecords = MemoryRecords.withRecords(
      2.toByte, 480L, Compression.NONE, TimestampType.CREATE_TIME, 123L, 0.toShort, 0, 0, false, localRecordList: _*
    )
    var localFileRecords: FileRecords = null

    // Diskless serves from the seal: offsets [1000, 1005). Models the control plane returning the
    // first batch at-or-after the requested (sub-seal) offset of 500.
    val supplementRecordList = (0 until 5).map(i => new SimpleRecord(i.toLong, s"diskless-$i".getBytes())).toArray
    val supplementRecords = MemoryRecords.withRecords(
      2.toByte, seal, Compression.NONE, TimestampType.CREATE_TIME, 456L, 0.toShort, 0, 0, false, supplementRecordList: _*
    )
    val disklessResponse = Map(disklessTopicPartition ->
      new FetchPartitionData(
        Errors.NONE,
        localLeo, 0L,
        supplementRecords,
        Optional.empty(), OptionalLong.of(localLeo), Optional.empty(), OptionalInt.empty(), false)
    )

    val fetchHandlerCtor = mockFetchHandler(disklessResponse)
    val cp = mock(classOf[ControlPlane])
    val replicaManager = spy(createReplicaManager(
      List(disklessTopicPartition.topic()),
      controlPlane = Some(cp),
      disklessManagedReplicasEnabled = true,
      disklessRemoteStorageConsolidationEnabled = true,
      consolidatingDisklessTopics = Set(disklessTopicPartition.topic()),
    ))
    try {
      when(replicaManager.inklessMetadataView().getClassicToDisklessStartOffset(disklessTopicPartition.topicPartition()))
        .thenReturn(seal)
      stubConsolidatingPartitionWithLocalLeo(replicaManager, localLeo = localLeo)

      localFileRecords = memoryRecordsToFileRecords(localMemRecords)
      doReturn(Seq(disklessTopicPartition ->
        new LogReadResult(
          new FetchDataInfo(new LogOffsetMetadata(480L, 0L, 0), localFileRecords),
          Optional.empty(), localLeo, 0L, localLeo, 0L, 0L, OptionalLong.empty(), Errors.NONE
        ))
      ).when(replicaManager).readFromLog(any(), any(), any(), any())

      val fetchParams = new FetchParams(
        FetchRequest.ORDINARY_CONSUMER_ID, -1L,
        1000L,
        localMemRecords.sizeInBytes + 1, // minBytes > local-only size to trigger the supplement
        1024 * 1024,
        FetchIsolation.HIGH_WATERMARK, Optional.empty()
      )
      val fetchInfos = Seq(
        disklessTopicPartition -> new PartitionData(disklessTopicPartition.topicId(), 480L, 0L, 1024 * 1024, Optional.empty())
      )

      @volatile var responseData: Map[TopicIdPartition, FetchPartitionData] = null
      val responseCallback = (response: Seq[(TopicIdPartition, FetchPartitionData)]) => {
        responseData = response.toMap
      }
      replicaManager.fetchMessages(fetchParams, fetchInfos, QuotaFactory.UNBOUNDED_QUOTA, responseCallback)

      waitForFetchResponse(responseData)
      val result = responseData(disklessTopicPartition)
      assertEquals(Errors.NONE, result.error)

      val offsets = result.records.records().asScala.map(_.offset()).toList
      val gaps = offsets.sliding(2).collect { case Seq(a, b) if b != a + 1 => (a, b) }.toList
      assertTrue(gaps.isEmpty,
        s"Consumer-visible records must be contiguous, but found offset gap(s): $gaps. " +
          s"The supplement started at offset 500 (below the seal $seal); diskless served from the " +
          s"seal, so the committed range [500, $seal) was silently skipped. Offsets=$offsets")
    } finally {
      replicaManager.shutdown(checkpointHW = false)
      fetchHandlerCtor.close()
      if (localFileRecords != null) localFileRecords.close()
    }
  }