Copilot commented on code in PR #23086:
URL: https://github.com/apache/kafka/pull/23086#discussion_r3770994702
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java:
##########
@@ -1533,6 +1535,596 @@ private void addRecords(final long messages, final
TopicPartition topicPartition
}
}
+ /**
+ * The first poll after assignment returns nothing and every record
arrives from the next poll
+ * on. An empty poll only means the fetch has not landed, so giving up on
it sends partitions
+ * to a log-start seek with no margin against retention.
+ */
+ @Test
+ public void shouldRetryProbePollBeforeFallingBackToLogStart() {
+ final long shortRetentionMs = Duration.ofSeconds(3).toMillis();
+ final long beginOffset = 900_000L;
+ final long logEndOffset = 1_000_000L;
+ final long seekTarget = 999_000L;
+ final int numPartitions = 12; // more than one poll can plausibly
serve at once
+
+ final TopicPartition[] tps = new TopicPartition[numPartitions];
+ final Map<TopicPartition, Long> begins = new HashMap<>();
+ final Map<TopicPartition, Long> ends = new HashMap<>();
+ for (int i = 0; i < numPartitions; i++) {
+ tps[i] = new TopicPartition(tp.topic(), i);
+ begins.put(tps[i], beginOffset);
+ ends.put(tps[i], logEndOffset);
+ }
+
+ // a position after restore reflects records since consumed, not where
it was seeked, so
+ // seekToBeginning is the only unambiguous signal that the
optimisation was abandoned
+ final Set<TopicPartition> seekedToBeginning = new HashSet<>();
+ final MockConsumer<byte[], byte[]> probeConsumer =
+ new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name()) {
+ @Override
+ public synchronized Map<TopicPartition, OffsetAndTimestamp>
offsetsForTimes(
+ final Map<TopicPartition, Long> timestampsToSearch) {
+ final Map<TopicPartition, OffsetAndTimestamp> result = new
HashMap<>();
+ timestampsToSearch.forEach((k, v) ->
+ result.put(k, new OffsetAndTimestamp(seekTarget, v)));
+ return result;
+ }
+
+ @Override
+ public synchronized void seekToBeginning(final
Collection<TopicPartition> partitions) {
+ seekedToBeginning.addAll(partitions);
+ super.seekToBeginning(partitions);
+ }
+ };
+ probeConsumer.updateBeginningOffsets(begins);
+ probeConsumer.updateEndOffsets(ends);
+ adminClient.updateEndOffsets(ends);
+
+ // records can only be added once assigned, and earlier polls happen
before that; the first
+ // poll after assignment delivers nothing, standing in for a fetch
that has not landed
+ final int[] assignedPolls = {0};
+ for (int round = 0; round < numPartitions * 4; round++) {
+ probeConsumer.schedulePollTask(() -> {
+ if (!probeConsumer.assignment().contains(tps[0])) {
+ return;
+ }
+ if (++assignedPolls[0] <= 1) {
+ return;
+ }
+ for (final TopicPartition partition : tps) {
+ probeConsumer.addRecord(new ConsumerRecord<>(
+ partition.topic(), partition.partition(), logEndOffset
- 1,
+ 10_000_000L, TimestampType.CREATE_TIME,
+ 0, 0, new byte[0], new byte[0], new RecordHeaders(),
Optional.empty()));
+ }
+ });
+ }
+
+ final StoreChangelogReader probeReader = new StoreChangelogReader(
+ time, config, logContext, adminClient, probeConsumer, callback,
standbyListener);
+
+ for (int i = 0; i < numPartitions; i++) {
+ final StateStoreMetadata meta = mock(StateStoreMetadata.class);
+ final ProcessorStateManager manager =
mock(ProcessorStateManager.class);
+ final StateStore store = mock(StateStore.class);
+ when(meta.changelogPartition()).thenReturn(tps[i]);
+ when(meta.store()).thenReturn(store);
+ when(meta.offset()).thenReturn(null, 0L); // no checkpoint, then
a value once restoring // no checkpoint
Review Comment:
This line has a duplicated trailing comment ("// no checkpoint") which adds
noise and can trip style checks/readability.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
