Skip to content

Commit 9f0a2f7

Browse files
authored
[server] Fix spammy logging when reporting backup version's replica ingestion timeout (linkedin#2359)
This PR addresses a spammy log issue when replica bootstrap timeout happens. Previously, when a push timed out, SIT would throw an exception unconditionally and stop consumption for all replicas. In a previous PR (linkedin#2284), when a backup or current version replica bootstrap times out, SIT no longer throws an exception but instead reports ERROR to PCS. For current version replicas, a reset feature may kick in to configure a Helix reset. For backup version replicas, such a feature is not triggered, and as a result, each check loop repeatedly reports the same log, which is not expected. We may consider extending the error reset feature to backup versions once it is stabilized, but this fix is still needed regardless. This PR resolves the issue by checking whether PCS has already reported ERROR. For replicas that have timed out and already reported ERROR, it will no longer log repeatedly.
1 parent 034453e commit 9f0a2f7

1 file changed

Lines changed: 4 additions & 4 deletions

File tree

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -622,8 +622,9 @@ protected void checkLongRunningTaskState() throws InterruptedException {
622622
* For current / backup version re-ingestion, it should report failure to the replica, but should keep other
623623
* online replica continue serving and do not close ingestion task.
624624
*/
625-
if (!partitionConsumptionState.isComplete() && LatencyUtils.getElapsedTimeFromMsToMs(
626-
partitionConsumptionState.getConsumptionStartTimeInMs()) > getBootstrapTimeoutInMs()) {
625+
if (!partitionConsumptionState.isComplete() && !partitionConsumptionState.isErrorReported()
626+
&& LatencyUtils.getElapsedTimeFromMsToMs(
627+
partitionConsumptionState.getConsumptionStartTimeInMs()) > getBootstrapTimeoutInMs()) {
627628
if (!pushTimeout) {
628629
pushTimeout = true;
629630
timeoutPartitions = new HashSet<>();
@@ -785,12 +786,11 @@ protected void checkLongRunningTaskState() throws InterruptedException {
785786
// Timeout
786787
String errorMsg =
787788
"After waiting " + TimeUnit.MILLISECONDS.toHours(getBootstrapTimeoutInMs()) + " hours, resource:"
788-
+ getStoreName() + " partitions:" + timeoutPartitions + " still can not complete ingestion.";
789+
+ getKafkaVersionTopic() + " partitions:" + timeoutPartitions + " still can not complete ingestion.";
789790
LOGGER.error(errorMsg);
790791
VeniceException ex = new VeniceTimeoutException(errorMsg);
791792
Store store = getStoreRepository().getStoreOrThrow(getStoreName());
792793
int currentVersion = store.getCurrentVersion();
793-
LOGGER.info("DEBUGGING: {} {}", currentVersion, getVersionNumber());
794794
if (getVersionNumber() <= currentVersion) {
795795
// For current / backup version, a replica's re-bootstrap timeout should not incur whole SIT closure.
796796
for (int partition: timeoutPartitions) {

0 commit comments

Comments
 (0)