[server] Add host-level signals for adaptive ingestion throttler #4014

Annotations

10 errors and 1 warning

Publish Test Report: clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/SITWithPWiseWithoutBufferAfterLeaderTest.java#L1

org.mockito.exceptions.verification.ArgumentsAreDifferent: Argument(s) are different! Wanted: storageMetadataService.put( "TestTopic_64c186d70d_1cecc12f_v1", 2, OffsetRecord{localVersionTopicOffset=3, upstreamOffset=-1, leaderTopic=null, offsetLag=0, eventTimeEpochMs=-1, latestProducerProcessingTimeInMs=0, isEndOfPushReceived=true, databaseInfo={}, realTimeProducerState={}} ); -> at com.linkedin.davinci.kafka.consumer.StoreIngestionTaskTest.lambda$testNotifier$35(StoreIngestionTaskTest.java:1623) Actual invocations have different arguments: storageMetadataService.getLastOffset( "TestTopic_64c186d70d_1cecc12f_v1", 1 ); -> at com.linkedin.venice.offsets.DeepCopyOffsetManager.getLastOffset(DeepCopyOffsetManager.java:49) storageMetadataService.getLastOffset( "TestTopic_64c186d70d_1cecc12f_v1", 2 ); -> at com.linkedin.venice.offsets.DeepCopyOffsetManager.getLastOffset(DeepCopyOffsetManager.java:49) storageMetadataService.computeStoreVersionState( "TestTopic_64c186d70d_1cecc12f_v1", com.linkedin.venice.offsets.DeepCopyStorageMetadataService$$Lambda$826/0x00007efdac6d3a48@2fd40a2e ); -> at com.linkedin.venice.offsets.DeepCopyStorageMetadataService.computeStoreVersionState(DeepCopyStorageMetadataService.java:33) storageMetadataService.put( "TestTopic_64c186d70d_1cecc12f_v1", 1, OffsetRecord{localVersionTopicOffset=1, upstreamOffset=-1, leaderTopic=null, offsetLag=9223372036854775807, eventTimeEpochMs=1737058537290, latestProducerProcessingTimeInMs=0, isEndOfPushReceived=false, databaseInfo={}, realTimeProducerState={}} ); -> at com.linkedin.venice.offsets.DeepCopyOffsetManager.put(DeepCopyOffsetManager.java:38) storageMetadataService.put( "TestTopic_64c186d70d_1cecc12f_v1", 1, OffsetRecord{localVersionTopicOffset=2, upstreamOffset=-1, leaderTopic=null, offsetLag=9223372036854775807, eventTimeEpochMs=1737058537292, latestProducerProcessingTimeInMs=0, isEndOfPushReceived=false, databaseInfo={}, realTimeProducerState={}} ); -> at com.linkedin.venice.offsets.DeepCopyOffsetManager.put(DeepCopyOffsetManager.java:38) storageMetadataService.computeStoreVersionState( "TestTopic_64c186d70d_1cecc12f_v1", com.linkedin.venice.offsets.DeepCopyStorageMetadataService$$Lambda$826/0x00007efdac6d3a48@1fcbc5b1 ); -> at com.linkedin.venice.offsets.DeepCopyStorageMetadataService.computeStoreVersionState(DeepCopyStorageMetadataService.java:33) storageMetadataService.put( "TestTopic_64c186d70d_1cecc12f_v1", 1, OffsetRecord{localVersionTopicOffset=3, upstreamOffset=-1, leaderTopic=null, offsetLag=9223372036854775807, eventTimeEpochMs=1737058537292, latestProducerProcessingTimeInMs=0, isEndOfPushReceived=true, databaseInfo={}, realTimeProducerState={}} ); -> at com.linkedin.venice.offsets.DeepCopyOffsetManager.put(DeepCopyOffsetManager.java:38) storageMetadataService.computeStoreVersionState( "TestTopic_64c186d70d_1cecc12f_v1", com.linkedin.venice.offsets.DeepCopyStorageMetadataService$$Lambda$826/0x00007efdac6d3a48@2aa15cac ); -> at com.linkedin.venice.offsets.DeepCopyStorageMetadataService.computeStoreVersionState(DeepCopyStorageMetadataService.java:33) storageMetadataService.put( "TestTopic_64c186d70d_1cecc12f_v1", 2, OffsetRecord{localVersionTopicOffset=1, upstreamOffset=-1, leaderTopic=null, offsetLag=9223372036854775807, eventTimeEpochMs=1737058537290, latestProducerProcessingTimeInMs=0, isEndOfPushReceived=false, databaseInfo={}, realTimeProducerState={}} ); -> at com.linkedin.venice.offsets.DeepCopyOffsetManager.put(DeepCopyOffsetManager.java:38) storageMetadataService.put( "TestTopic_64c186d70d_1cecc12f_v1", 2, OffsetRecord{localVersionTopicOffset=2, upstreamOffset=-1, leaderTopic=null, offsetLag=9223372036854775807, eventTimeEpochMs=1737058537292, latestProducerProcessingTimeInMs=0, isEndOfPushReceived=false, databaseInfo={}, realTimeProducerState={}} ); -> at com.linkedin.venice.offsets.DeepCopyOffsetManager.put(DeepCopyOffsetManager.java:38) storageMetadataService.computeStoreVersionState( "TestTopic_64c186d70d_1cecc12f_v1",

Publish Test Report: clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/SITWithPWiseWithoutBufferAfterLeaderTest.java#L1

org.mockito.exceptions.verification.TooFewActualInvocations: activeActiveStoreIngestionTask.resubscribeForAllPartitions(); Wanted 4 times: -> at com.linkedin.davinci.kafka.consumer.StoreIngestionTask.resubscribeForAllPartitions(StoreIngestionTask.java:630) But was 3 times: -> at com.linkedin.davinci.kafka.consumer.StoreIngestionTask.updateIngestionRoleIfStoreChanged(StoreIngestionTask.java:1570) -> at com.linkedin.davinci.kafka.consumer.StoreIngestionTask.updateIngestionRoleIfStoreChanged(StoreIngestionTask.java:1570) -> at com.linkedin.davinci.kafka.consumer.StoreIngestionTask.updateIngestionRoleIfStoreChanged(StoreIngestionTask.java:1570)

Publish Test Report: clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/SITWithTWiseWithoutBufferAfterLeaderTest.java#L1

org.mockito.exceptions.verification.ArgumentsAreDifferent: Argument(s) are different! Wanted: storageMetadataService.put( "TestTopic_683183d2a3_7ffee30_v1", 2, OffsetRecord{localVersionTopicOffset=3, upstreamOffset=-1, leaderTopic=null, offsetLag=0, eventTimeEpochMs=-1, latestProducerProcessingTimeInMs=0, isEndOfPushReceived=true, databaseInfo={}, realTimeProducerState={}} ); -> at com.linkedin.davinci.kafka.consumer.StoreIngestionTaskTest.lambda$testNotifier$35(StoreIngestionTaskTest.java:1623) Actual invocations have different arguments: storageMetadataService.getLastOffset( "TestTopic_683183d2a3_7ffee30_v1", 1 ); -> at com.linkedin.venice.offsets.DeepCopyOffsetManager.getLastOffset(DeepCopyOffsetManager.java:49) storageMetadataService.getLastOffset( "TestTopic_683183d2a3_7ffee30_v1", 2 ); -> at com.linkedin.venice.offsets.DeepCopyOffsetManager.getLastOffset(DeepCopyOffsetManager.java:49) storageMetadataService.computeStoreVersionState( "TestTopic_683183d2a3_7ffee30_v1", com.linkedin.venice.offsets.DeepCopyStorageMetadataService$$Lambda$1021/0x00007f087074c470@611158a5 ); -> at com.linkedin.venice.offsets.DeepCopyStorageMetadataService.computeStoreVersionState(DeepCopyStorageMetadataService.java:33) storageMetadataService.computeStoreVersionState( "TestTopic_683183d2a3_7ffee30_v1", com.linkedin.venice.offsets.DeepCopyStorageMetadataService$$Lambda$1021/0x00007f087074c470@391b3770 ); -> at com.linkedin.venice.offsets.DeepCopyStorageMetadataService.computeStoreVersionState(DeepCopyStorageMetadataService.java:33) storageMetadataService.put( "TestTopic_683183d2a3_7ffee30_v1", 1, OffsetRecord{localVersionTopicOffset=1, upstreamOffset=-1, leaderTopic=null, offsetLag=9223372036854775807, eventTimeEpochMs=1737058552056, latestProducerProcessingTimeInMs=0, isEndOfPushReceived=false, databaseInfo={}, realTimeProducerState={}} ); -> at com.linkedin.venice.offsets.DeepCopyOffsetManager.put(DeepCopyOffsetManager.java:38) storageMetadataService.put( "TestTopic_683183d2a3_7ffee30_v1", 1, OffsetRecord{localVersionTopicOffset=2, upstreamOffset=-1, leaderTopic=null, offsetLag=9223372036854775807, eventTimeEpochMs=1737058552057, latestProducerProcessingTimeInMs=0, isEndOfPushReceived=false, databaseInfo={}, realTimeProducerState={}} ); -> at com.linkedin.venice.offsets.DeepCopyOffsetManager.put(DeepCopyOffsetManager.java:38) storageMetadataService.computeStoreVersionState( "TestTopic_683183d2a3_7ffee30_v1", com.linkedin.venice.offsets.DeepCopyStorageMetadataService$$Lambda$1021/0x00007f087074c470@9e4f756 ); -> at com.linkedin.venice.offsets.DeepCopyStorageMetadataService.computeStoreVersionState(DeepCopyStorageMetadataService.java:33) storageMetadataService.put( "TestTopic_683183d2a3_7ffee30_v1", 2, OffsetRecord{localVersionTopicOffset=1, upstreamOffset=-1, leaderTopic=null, offsetLag=9223372036854775807, eventTimeEpochMs=1737058552056, latestProducerProcessingTimeInMs=0, isEndOfPushReceived=false, databaseInfo={}, realTimeProducerState={}} ); -> at com.linkedin.venice.offsets.DeepCopyOffsetManager.put(DeepCopyOffsetManager.java:38) storageMetadataService.put( "TestTopic_683183d2a3_7ffee30_v1", 2, OffsetRecord{localVersionTopicOffset=2, upstreamOffset=-1, leaderTopic=null, offsetLag=9223372036854775807, eventTimeEpochMs=1737058552057, latestProducerProcessingTimeInMs=0, isEndOfPushReceived=false, databaseInfo={}, realTimeProducerState={}} ); -> at com.linkedin.venice.offsets.DeepCopyOffsetManager.put(DeepCopyOffsetManager.java:38) storageMetadataService.computeStoreVersionState( "TestTopic_683183d2a3_7ffee30_v1", com.linkedin.venice.offsets.DeepCopyStorageMetadataService$$Lambda$1021/0x00007f087074c470@18ccae91 ); -> at com.linkedin.venice.offsets.DeepCopyStorageMetadataService.computeStoreVersionState(DeepCopyStorageMetadataService.java:33) storageMetadataService.put( "TestTopic_683183d2a3_7ffee30_v1", 1, OffsetRecord{localVersionTopicOffset=3, upstreamOffset=-1, leaderTopic=null, offsetLag=9223372036854775807, eventTi

Publish Test Report: clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/TestRemoteIngestionRepairService.java#L25

java.lang.AssertionError: expected [2] but found [1]

Publish Test Report: clients/da-vinci-client/src/test/java/com/linkedin/davinci/repository/NativeMetadataRepositoryTest.java#L133

java.lang.AssertionError: expected [2000.0] but found [1000.0]

Publish Test Report: clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/SITWithPWiseWithoutBufferAfterLeaderTest.java#L1