unknowntpo commented on code in PR #22309:
URL: https://github.com/apache/kafka/pull/22309#discussion_r3314791637
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java:
##########
@@ -1562,6 +1563,117 @@ public void
shouldSeekToBeginningForNonWindowedStoreWithoutCheckpoint() {
assertEquals(0L, consumer.position(tp), "Non-windowed store should
seek to beginning, not by timestamp");
}
+ @Test
+ public void shouldRejectInvalidRestoreBufferedRecordsPerPartition() {
+ final Properties properties = new Properties();
+
properties.put(StreamsConfig.RESTORE_BUFFERED_RECORDS_PER_PARTITION_CONFIG, 0);
+ assertThrows(ConfigException.class,
+ () -> new
StreamsConfig(StreamsTestUtils.getStreamsConfig("test-reader", properties)));
+ }
+
+ @Test
+ public void shouldPausePartitionWhenRestoreBufferCapIsReached() {
+ // standby with a committed-offset limit lets records pile up past the
limit
+ setupStandbyStateManager();
+ setupStoreMetadata();
+ setupStore();
+ // standby fixture has a null taskId, so the predicate looks up
tasks.get(null)
+ @SuppressWarnings("unchecked")
+ final Map<TaskId, Task> mockTasks = mock(Map.class);
+ when(mockTasks.get(null)).thenReturn(mock(Task.class));
+ when(mockTasks.containsKey(null)).thenReturn(true);
+ when(standbyStateManager.changelogAsSource(tp)).thenReturn(true);
+ when(storeMetadata.offset()).thenReturn(3L);
+ when(storeMetadata.endOffset()).thenReturn(20L);
+
+ final Properties properties = new Properties();
+
properties.put(StreamsConfig.RESTORE_BUFFERED_RECORDS_PER_PARTITION_CONFIG, 3);
+ properties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
+ final StreamsConfig cappedConfig =
+ new StreamsConfig(StreamsTestUtils.getStreamsConfig("test-reader",
properties));
+ final StoreChangelogReader changelogReader =
+ new StoreChangelogReader(time, cappedConfig, logContext,
adminClient, consumer, callback, standbyListener);
+ changelogReader.transitToUpdateStandby();
+
+ consumer.updateBeginningOffsets(Collections.singletonMap(tp, 0L));
+ adminClient.updateConsumerGroupOffsets(Collections.singletonMap(tp,
7L));
+ changelogReader.register(tp, standbyStateManager);
+
+ changelogReader.restore(mockTasks);
+ assertEquals(Collections.emptySet(), consumer.paused());
+
+ // committed offset is 7, so offsets >=7 stay buffered
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 5L,
"key".getBytes(), "value".getBytes()));
Review Comment:
nit: Can we use `tp.partition()` here to keep the record partition tied to
the `TopicPartition` used by the test?
```suggestion
consumer.addRecord(new ConsumerRecord<>(topicName, tp.partition(),
5L, "key".getBytes(), "value".getBytes()));
```
--
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]
