unknowntpo commented on code in PR #22309:
URL: https://github.com/apache/kafka/pull/22309#discussion_r3314898327
##########
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()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 6L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 7L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 8L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 9L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 10L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 11L,
"key".getBytes(), "value".getBytes()));
+
+ // first tick fills the buffer, second tick sees it full and pauses
+ changelogReader.restore(mockTasks);
+ changelogReader.restore(mockTasks);
+
+
assertTrue(changelogReader.changelogMetadata(tp).bufferedRecords().size() >= 3);
Review Comment:
nit: Could we use a named local variable for
`RESTORE_BUFFERED_RECORDS_PER_PARTITION_CONFIG` here, and assert against it
instead of hard-coded `3`? Its more readable.
```suggestion
final int restoreBufferedRecordsPerPartition = 3;
properties.put(StreamsConfig.RESTORE_BUFFERED_RECORDS_PER_PARTITION_CONFIG,
restoreBufferedRecordsPerPartition);
```
Then for the assertion line:
```suggestion
assertTrue(changelogReader.changelogMetadata(tp).bufferedRecords().size() >=
restoreBufferedRecordsPerPartition);
```
--
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]
