Repository: apache/kafka
File: core/src/test/java/kafka/server/share/SharePartitionManagerTest.java (Java)

Code before:
            Option.empty(),
            Option.empty(),
            Option.empty()
        ))));
        return CollectionConverters.asScala(logReadResults).toSeq();
    }
    static Seq<Tuple2<TopicIdPartition, LogReadResult>> buildLogReadResultWithFakeRecords(List<TopicIdPartition> topicIdPartitions) {
        List<Tuple2<TopicIdPartition, LogReadResult>> logReadResults = new ArrayList<>();
        for (TopicIdPartition topicIdPartition : topicIdPartitions) {
            MemoryRecords records = MemoryRecords.withRecords(
                    Compression.NONE,
                    new SimpleRecord("test-key".getBytes(), "test-value".getBytes())
            );

            LogReadResult logReadResult = new LogReadResult(
                    new FetchDataInfo(new LogOffsetMetadata(0, 0, 0), records),
                    Option.empty(),
                    -1L, -1L, -1L, -1L, -1L,
                    Option.empty(), Option.empty(), Option.empty()
            );

            logReadResults.add(new Tuple2<>(topicIdPartition, logReadResult));
        }

        return CollectionConverters.asScala(logReadResults).toSeq();
    }

    static void mockReplicaManagerDelayedShareFetch(ReplicaManager replicaManager,
                                                    DelayedOperationPurgatory<DelayedShareFetch> delayedShareFetchPurgatory) {
        doAnswer(invocationOnMock -> {
            Object[] args = invocationOnMock.getArguments();
            DelayedShareFetchKey key = (DelayedShareFetchKey) args[0];
            delayedShareFetchPurgatory.checkAndComplete(key);
            return null;
        }).when(replicaManager).completeDelayedShareFetchRequest(any(DelayedShareFetchKey.class));

        doAnswer(invocationOnMock -> {
            Object[] args = invocationOnMock.getArguments();
            DelayedShareFetch operation = (DelayedShareFetch) args[0];
            List<DelayedOperationKey> keys = (List<DelayedOperationKey>) args[1];
            delayedShareFetchPurgatory.tryCompleteElseWatch(operation, keys);
            return null;
        }).when(replicaManager).addDelayedShareFetchRequest(any(), any());
    }

    private record TopicMetrics(
        long totalShareFetchRequestCount,
        long failedShareFetchRequestCount,
        long totalShareAcknowledgementRequestCount,
        long failedShareAcknowledgementRequestCount
    ) { }

Review comment (suggestion):
You can make the change from `MemoryRecords.EMPTY` to `MemoryRecords.withRecords(Compression.NONE, new SimpleRecord("test-key".getBytes(), "test-value".getBytes()))` in the function `buildLogReadResult` itself.

Code after:
            Option.empty(),
            Option.empty()
        ))));
        return CollectionConverters.asScala(logReadResults).toSeq();
    }

    static void mockReplicaManagerDelayedShareFetch(ReplicaManager replicaManager,
                                                    DelayedOperationPurgatory<DelayedShareFetch> delayedShareFetchPurgatory) {
        doAnswer(invocationOnMock -> {
            Object[] args = invocationOnMock.getArguments();
            DelayedShareFetchKey key = (DelayedShareFetchKey) args[0];
            delayedShareFetchPurgatory.checkAndComplete(key);
            return null;
        }).when(replicaManager).completeDelayedShareFetchRequest(any(DelayedShareFetchKey.class));

        doAnswer(invocationOnMock -> {
            Object[] args = invocationOnMock.getArguments();
            DelayedShareFetch operation = (DelayedShareFetch) args[0];
            List<DelayedOperationKey> keys = (List<DelayedOperationKey>) args[1];
            delayedShareFetchPurgatory.tryCompleteElseWatch(operation, keys);
            return null;
        }).when(replicaManager).addDelayedShareFetchRequest(any(), any());
    }

    private record TopicMetrics(
        long totalShareFetchRequestCount,
        long failedShareFetchRequestCount,
        long totalShareAcknowledgementRequestCount,
        long failedShareAcknowledgementRequestCount
    ) { }

    static class SharePartitionManagerBuilder {
        private final Persister persister = new NoOpStatePersister();
        private ReplicaManager replicaManager = mock(ReplicaManager.class);
        private Time time = new MockTime();
        private ShareSessionCache cache = new ShareSessionCache(10);
        private SharePartitionCache partitionCache = new SharePartitionCache();
        private Timer timer = new MockTimer();
        private ShareGroupMetrics shareGroupMetrics = new ShareGroupMetrics(time);
        private BrokerTopicStats brokerTopicStats;

        private SharePartitionManagerBuilder withReplicaManager(ReplicaManager replicaManager) {
            this.replicaManager = replicaManager;
            return this;
        }

        private SharePartitionManagerBuilder withTime(Time time) {
            this.time = time;
            return this;
        }