(hudi) branch master updated: chore(utilities): add SQS backlog / in-flight visibility to S3 events source (#19333)

[vhs]

This is an automated email from the ASF dual-hosted git repository.

voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git


The following commit(s) were added to refs/heads/master by this push:
     new e4071ce54076 chore(utilities): add SQS backlog / in-flight visibility 
to S3 events source (#19333)
e4071ce54076 is described below

commit e4071ce5407656e6acd6214b77fcc66813bce69b
Author: Davis-Zhang-Onehouse 
<169106455+davis-zhang-oneho...@users.noreply.github.com>
AuthorDate: Wed Jul 22 03:35:14 2026 -0700

    chore(utilities): add SQS backlog / in-flight visibility to S3 events 
source (#19333)
    
    * logging: add SQS backlog / in-flight visibility to S3 events source
    
    Adds a lifecycle log line to S3EventsSource.onCommit recording how many SQS
    messages are being deleted at commit and the checkpoint, improving 
operational
    visibility into the SQS-backed S3 events ingestion path.
    
    Placeholder / first increment of a broader observability change tracked for 
an
    Onehouse internal hotfix; the full receive-loop / in-flight logging will be
    upstreamed here subsequently.
    
    * chore(utilities): drop redundant class/method prefix from onCommit log
    
    SLF4J emits the logger name already (every log4j2 pattern in the repo uses
    %c), so the "S3EventsSource.onCommit:" prefix duplicated it. Also names the
    queue as SQS in the message.
    
    ---------
    
    Co-authored-by: voon <voonho...@gmail.com>
---
 .../main/java/org/apache/hudi/utilities/sources/S3EventsSource.java | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git 
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/S3EventsSource.java
 
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/S3EventsSource.java
index 6d96542d0bbb..b6b66d3fb418 100644
--- 
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/S3EventsSource.java
+++ 
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/S3EventsSource.java
@@ -35,6 +35,8 @@ import org.apache.spark.sql.Encoders;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.types.StructType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.services.sqs.SqsClient;
 
 import java.io.Closeable;
@@ -52,6 +54,8 @@ import static 
org.apache.hudi.utilities.config.CloudSourceConfig.META_EVENTS_PER
  */
 public class S3EventsSource extends RowSource implements Closeable {
 
+  private static final Logger LOG = 
LoggerFactory.getLogger(S3EventsSource.class);
+
   private final S3EventsMetaSelector pathSelector;
   private final SchemaProvider schemaProvider;
   private final List<CloudObjectsSelector.MessageTracker> processedMessages = 
new ArrayList<>();
@@ -109,6 +113,8 @@ public class S3EventsSource extends RowSource implements 
Closeable {
 
   @Override
   public void onCommit(String lastCkptStr) {
+    LOG.info("Deleting {} processed messages from SQS queue, checkpoint={}.",
+        processedMessages.size(), lastCkptStr);
     pathSelector.deleteProcessedMessages(sqs, pathSelector.queueUrl, 
processedMessages);
     processedMessages.clear();
   }