adixitconfluent commented on code in PR #19786:
URL: https://github.com/apache/kafka/pull/19786#discussion_r2103876879
##########
core/src/test/java/kafka/server/share/SharePartitionManagerTest.java:
##########
@@ -3158,6 +3180,26 @@ static Seq<Tuple2<TopicIdPartition, LogReadResult>>
buildLogReadResult(List<Topi
))));
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();
+ }
Review Comment:
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.
##########
core/src/test/java/kafka/server/share/SharePartitionManagerTest.java:
##########
@@ -1723,7 +1743,9 @@ public void
testAcknowledgeCompletesDelayedShareFetchRequest() {
// Since acquisition lock for sp1 and sp2 cannot be acquired, we
should have 2 watched keys.
assertEquals(2, delayedShareFetchPurgatory.watched());
- doAnswer(invocation ->
buildLogReadResult(topicIdPartitions)).when(mockReplicaManager).readFromLog(any(),
any(), any(ReplicaQuota.class), anyBoolean());
+ doAnswer(invocation -> buildLogReadResult(List.of(tp1)))
+ .when(mockReplicaManager)
+ .readFromLog(any(), any(), any(ReplicaQuota.class),
anyBoolean());
Review Comment:
nit: Extra tabs on both lines
##########
core/src/test/java/kafka/server/share/SharePartitionManagerTest.java:
##########
@@ -790,6 +789,27 @@ public void
testCachedTopicPartitionsForValidShareSessions() {
String groupId = "grp";
Uuid memberId1 = Uuid.randomUuid();
Uuid memberId2 = Uuid.randomUuid();
+ SharePartition sp0 = mock(SharePartition.class);
+ SharePartition sp1 = mock(SharePartition.class);
+ SharePartition sp2 = mock(SharePartition.class);
+ SharePartition sp3 = mock(SharePartition.class);
+
+
when(sp0.releaseAcquiredRecords(ArgumentMatchers.eq(String.valueOf(memberId1)))).thenReturn(CompletableFuture.completedFuture(null));
+
when(sp1.releaseAcquiredRecords(ArgumentMatchers.eq(String.valueOf(memberId1)))).thenReturn(CompletableFuture.completedFuture(null));
+
when(sp2.releaseAcquiredRecords(ArgumentMatchers.eq(String.valueOf(memberId1)))).thenReturn(CompletableFuture.completedFuture(null));
+
when(sp3.releaseAcquiredRecords(ArgumentMatchers.eq(String.valueOf(memberId1)))).thenReturn(CompletableFuture.completedFuture(null));
+
+ SharePartitionCache partitionCache = new SharePartitionCache();
+ partitionCache.put(new SharePartitionKey(groupId, tp0), sp0);
+ partitionCache.put(new SharePartitionKey(groupId, tp1), sp1);
+ partitionCache.put(new SharePartitionKey(groupId, tp2), sp2);
+ partitionCache.put(new SharePartitionKey(groupId, tp3), sp3);
Review Comment:
I don't think you need `sp3`. releaseSession is only called for tp0, tp1,
and tp2. So, just mocking these 3 should work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
