jolshan commented on code in PR #14417:
URL: https://github.com/apache/kafka/pull/14417#discussion_r1344714254
##########
core/src/test/scala/unit/kafka/coordinator/group/CoordinatorLoaderImplTest.scala:
##########
@@ -283,7 +292,63 @@ class CoordinatorLoaderImplTest {
minOneMessage = true
)).thenReturn(readResult)
- assertNull(loader.load(tp, coordinator).get(10, TimeUnit.SECONDS))
+ assertNotNull(loader.load(tp, coordinator).get(10, TimeUnit.SECONDS))
+ }
+ }
+
+ @Test
+ def testLoadSummary(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val serde = new StringKeyValueDeserializer
+ val log = mock(classOf[UnifiedLog])
+ val coordinator = mock(classOf[CoordinatorPlayback[(String, String)]])
+ val time = new MockTime()
+
+ TestUtils.resource(new CoordinatorLoaderImpl[(String, String)](
+ time,
+ replicaManager = replicaManager,
+ deserializer = serde,
+ loadBufferSize = 1000
+ )) { loader =>
+ val startTimeMs = time.milliseconds()
+ when(replicaManager.getLog(tp)).thenReturn(Some(log))
+ when(log.logStartOffset).thenReturn(0L)
+ when(replicaManager.getLogEndOffset(tp)).thenReturn(Some(5L))
+
+ val readResult1 = logReadResult(startOffset = 0, records = Seq(
+ new SimpleRecord("k1".getBytes, "v1".getBytes),
+ new SimpleRecord("k2".getBytes, "v2".getBytes)
+ ))
+
+ when(log.read(
+ startOffset = 0L,
+ maxLength = 1000,
+ isolation = FetchIsolation.LOG_END,
+ minOneMessage = true
+ )).thenAnswer((_: InvocationOnMock) => {
+ time.sleep(1000)
Review Comment:
I think the idea was that we would have a different value for the end time
for the check on line 349. I think this is ok.
--
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
