Repository: risingwavelabs/risingwave
File: src/stream/src/executor/source/fs_fetch_executor.rs (Rust)

Code before:
                            fragment_id = %self.actor_ctx.fragment_id,
                            reading_file = %split_id,
                            retries_done = *retries_done,
                            max_retries = MAX_RETRIES_PER_SPLIT,
                            error = %e.as_report(),
                            "Fetch Error, retrying file split"
                        );
                        tokio::time::sleep(backoff).await;
                    } else {
                        // Exceeded max retries: mark dirty in memory and skip this split afterwards.
                        dirty_splits.insert(split_id.clone());
                        retry_counts.remove(&split_id);
                        tracing::error!(
                            source_id = %core.source_id,
                            source_name = %core.source_name,
                            fragment_id = %self.actor_ctx.fragment_id,
                            reading_file = %split_id,
                            max_retries = MAX_RETRIES_PER_SPLIT,
                            error = %e.as_report(),
                            "Fetch Error, exceeded max retries; marking split dirty and skipping"
                        );
                        GLOBAL_ERROR_METRICS.user_source_error.report([
                            "File source dirty split".to_owned(),
                            core.source_id.to_string(),
                            core.source_name.clone(),
                            self.actor_ctx.fragment_id.to_string(),
                        ]);
                    }

                    // Clear current reading file and rebuild reader to continue with other splits.
                    *reading_file.lock().expect("mutex poisoned") = None;
                    splits_on_fetch = 0;
                    Self::replace_with_new_batch_reader(ReplaceReaderArgs {
                        splits_on_fetch: &mut splits_on_fetch,
                        state_store_handler: &state_store_handler,
                        dirty_splits: &dirty_splits,
                        column_ids: core.column_ids.clone(),
                        source_ctx: self.build_source_ctx(
                            &source_desc,
                            core.source_id,
                            &core.source_name,
                        ),
                        source_desc: &source_desc,
                        stream: &mut stream,
                        rate_limit_rps: self.rate_limit_rps,
                        reading_file: reading_file.clone(),
                    })
                    .await?;
                    continue;
                }
                Ok(msg) => {

Review comment (bug):
`retry_counts` only clean up when exceeded max retries, shall we remove also when the file failed but succeed?

Code after:
                        return Err(e);
                    };

                    let retries_done = retry_counts.entry(split_id.clone()).or_insert(0);
                    if *retries_done < MAX_RETRIES_PER_SPLIT {
                        *retries_done = retries_done.saturating_add(1);
                        let backoff = RETRY_BASE_BACKOFF
                            .checked_mul(1u32 << (*retries_done - 1))
                            .unwrap_or(Duration::from_secs(60));
                        tracing::warn!(
                            source_id = %core.source_id,
                            source_name = %core.source_name,
                            fragment_id = %self.actor_ctx.fragment_id,
                            reading_file = %split_id,
                            retries_done = *retries_done,
                            max_retries = MAX_RETRIES_PER_SPLIT,
                            error = %e.as_report(),
                            "Fetch Error, retrying file split"
                        );
                        tokio::time::sleep(backoff).await;
                    } else {
                        // Exceeded max retries: mark dirty in memory and skip this split afterwards.
                        dirty_splits.insert(split_id.clone());
                        dirty_split_count_metrics.set(dirty_splits.len() as i64);
                        failed_split_count_metrics.inc();
                        retry_counts.remove(&split_id);
                        tracing::error!(
                            source_id = %core.source_id,
                            source_name = %core.source_name,
                            fragment_id = %self.actor_ctx.fragment_id,
                            reading_file = %split_id,
                            max_retries = MAX_RETRIES_PER_SPLIT,
                            error = %e.as_report(),
                            "Fetch Error, exceeded max retries; marking split dirty and skipping"
                        );
                        GLOBAL_ERROR_METRICS.user_source_error.report([
                            "File source dirty split".to_owned(),
                            core.source_id.to_string(),
                            core.source_name.clone(),
                            self.actor_ctx.fragment_id.to_string(),
                        ]);
                    }

                    // Clear current reading file and rebuild reader to continue with other splits.
                    *reading_file.lock().expect("mutex poisoned") = None;
                    splits_on_fetch = 0;
                    Self::replace_with_new_batch_reader(ReplaceReaderArgs {
                        splits_on_fetch: &mut splits_on_fetch,
                        state_store_handler: &state_store_handler,
                        dirty_splits: &dirty_splits,
                        column_ids: core.column_ids.clone(),