From c439e61fae414f8c6e49390b0abf136a7860d080 Mon Sep 17 00:00:00 2001 From: kanthi subramanian Date: Sat, 15 Aug 2026 17:45:38 -0500 Subject: [PATCH 1/5] Support for buffering inserts --- examples/grafana/METRICS.md | 3 + ice-rest-catalog/pom.xml | 2 +- ice/pom.xml | 7 + .../main/java/com/altinity/ice/cli/Main.java | 27 +- .../ice/cli/internal/cmd/InsertWatch.java | 372 ++++++++++++++---- .../cli/internal/cmd/InsertWatchBuffer.java | 151 +++++++ .../internal/metrics/InsertWatchMetrics.java | 53 +++ pom.xml | 1 + 8 files changed, 537 insertions(+), 79 deletions(-) create mode 100644 ice/src/main/java/com/altinity/ice/cli/internal/cmd/InsertWatchBuffer.java diff --git a/examples/grafana/METRICS.md b/examples/grafana/METRICS.md index 4dc0287e..27f96413 100644 --- a/examples/grafana/METRICS.md +++ b/examples/grafana/METRICS.md @@ -120,6 +120,9 @@ These metrics track S3 event-driven file insertions. | `ice_watch_queue_receive_errors_total` | Counter | table, queue, queue_type | Total errors when receiving messages from queue | | `ice_watch_queue_delete_errors_total` | Counter | table, queue, queue_type | Total errors when deleting/acknowledging messages | | `ice_watch_message_parse_errors_total` | Counter | table, queue, queue_type | Total message parsing errors | +| `ice_watch_buffer_files` | Gauge | table, queue, queue_type | Files accumulated and waiting to be committed | +| `ice_watch_buffer_bytes` | Gauge | table, queue, queue_type | Total size of files accumulated and waiting to be committed | +| `ice_watch_buffer_flushes_total` | Counter | table, queue, queue_type, trigger | Accumulated batches committed, by the threshold that triggered the commit | ### Maintenance Metrics diff --git a/ice-rest-catalog/pom.xml b/ice-rest-catalog/pom.xml index 5cbd7be9..66288b7a 100644 --- a/ice-rest-catalog/pom.xml +++ b/ice-rest-catalog/pom.xml @@ -547,7 +547,7 @@ com.github.shyiko.skedule skedule - 0.4.0 + ${skedule.version} kalvanized diff --git a/ice/pom.xml b/ice/pom.xml index e4ed3fda..9a320f87 100644 --- a/ice/pom.xml +++ b/ice/pom.xml @@ -526,6 +526,13 @@ prometheus-metrics-exporter-servlet-jakarta ${prometheus.version} + + + com.github.shyiko.skedule + skedule + ${skedule.version} + kalvanized + info.picocli diff --git a/ice/src/main/java/com/altinity/ice/cli/Main.java b/ice/src/main/java/com/altinity/ice/cli/Main.java index 2c029488..a1494413 100644 --- a/ice/src/main/java/com/altinity/ice/cli/Main.java +++ b/ice/src/main/java/com/altinity/ice/cli/Main.java @@ -26,6 +26,7 @@ import com.altinity.ice.cli.internal.cmd.Files; import com.altinity.ice.cli.internal.cmd.Insert; import com.altinity.ice.cli.internal.cmd.InsertWatch; +import com.altinity.ice.cli.internal.cmd.InsertWatchBuffer; import com.altinity.ice.cli.internal.cmd.ListNamespaces; import com.altinity.ice.cli.internal.cmd.ListPartitions; import com.altinity.ice.cli.internal.cmd.ListSnapshots; @@ -557,7 +558,25 @@ void insert( @CommandLine.Option( names = {"--watch-debug-addr"}, description = "") - String watchDebugAddr) + String watchDebugAddr, + @CommandLine.Option( + names = {"--watch-commit-schedule"}, + description = + "Accumulate incoming files and commit them as a single snapshot on this schedule," + + " in https://github.com/shyiko/skedule format, e.g. \"every 5 minutes\"," + + " \"every day 02:00\" (default: commit on every poll)") + String watchCommitSchedule, + @CommandLine.Option( + names = {"--watch-max-files"}, + description = "Commit as soon as this many files are accumulated (default: no limit)", + defaultValue = "0") + int watchMaxFiles, + @CommandLine.Option( + names = {"--watch-max-bytes"}, + description = + "Commit as soon as accumulated files add up to this many bytes (default: no limit)", + defaultValue = "0") + long watchMaxBytes) throws IOException, InterruptedException { if (s3NoSignRequest && s3CopyObject) { throw new UnsupportedOperationException( @@ -606,6 +625,11 @@ void insert( TableIdentifier tableId = TableIdentifier.parse(name); boolean watchMode = !Strings.isNullOrEmpty(watch); + if (!watchMode && (watchCommitSchedule != null || watchMaxFiles > 0 || watchMaxBytes > 0)) { + throw new IllegalArgumentException( + "--watch-commit-schedule/--watch-max-files/--watch-max-bytes require --watch"); + } + if (createTableIfNotExists && !watchMode) { CreateTable.run( catalog, @@ -679,6 +703,7 @@ void insert( watchFireOnce, createTableIfNotExists, options, + new InsertWatchBuffer.BatchOptions(watchCommitSchedule, watchMaxFiles, watchMaxBytes), metricsEnabled); } } diff --git a/ice/src/main/java/com/altinity/ice/cli/internal/cmd/InsertWatch.java b/ice/src/main/java/com/altinity/ice/cli/internal/cmd/InsertWatch.java index 4dfcf15e..3aa9afe1 100644 --- a/ice/src/main/java/com/altinity/ice/cli/internal/cmd/InsertWatch.java +++ b/ice/src/main/java/com/altinity/ice/cli/internal/cmd/InsertWatch.java @@ -9,21 +9,27 @@ */ package com.altinity.ice.cli.internal.cmd; +import com.altinity.ice.cli.internal.cmd.InsertWatchBuffer.BatchOptions; +import com.altinity.ice.cli.internal.cmd.InsertWatchBuffer.FilterResult; import com.altinity.ice.cli.internal.metrics.InsertWatchMetrics; import com.altinity.ice.internal.io.Matcher; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.github.shyiko.skedule.Schedule; import java.io.IOException; import java.net.URI; import java.net.URLDecoder; import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.time.ZonedDateTime; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; -import java.util.LinkedHashSet; +import java.util.LinkedHashMap; import java.util.LinkedList; import java.util.List; +import java.util.Map; import java.util.function.Supplier; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.exceptions.NoSuchTableException; @@ -35,6 +41,9 @@ import software.amazon.awssdk.services.sqs.SqsClient; import software.amazon.awssdk.services.sqs.SqsClientBuilder; import software.amazon.awssdk.services.sqs.model.BatchResultErrorEntry; +import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityBatchRequest; +import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityBatchRequestEntry; +import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityBatchResponse; import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequest; import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequestEntry; import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchResponse; @@ -47,6 +56,13 @@ public class InsertWatch { private static final ObjectMapper objectMapper = new ObjectMapper(); private static final String QUEUE_TYPE_SQS = "sqs"; + // Maximum number of entries accepted by the SQS batch APIs. + private static final int SQS_BATCH_LIMIT = 10; + + // Floor for how long accumulated messages are kept invisible. + private static final int MIN_VISIBILITY_TIMEOUT_SECONDS = 60; + private static final int MAX_VISIBILITY_TIMEOUT_SECONDS = 43200; // SQS limit (12h) + public static void run( RESTCatalog catalog, TableIdentifier nsTable, @@ -65,6 +81,7 @@ public static void run( terminateAfterOneBatch, createTableIfNotExists, options, + BatchOptions.NONE, false); } @@ -77,6 +94,7 @@ public static void run( boolean terminateAfterOneBatch, boolean createTableIfNotExists, Insert.Options options, + BatchOptions batchOptions, boolean metricsEnabled) throws IOException, InterruptedException { @@ -98,6 +116,25 @@ public static void run( String queueLabel = sqsQueueURL; String queueType = QUEUE_TYPE_SQS; + Schedule schedule = + batchOptions.commitSchedule() != null + ? Schedule.parse(batchOptions.commitSchedule()) + : null; + ZonedDateTime nextCommitAt = schedule != null ? schedule.next(ZonedDateTime.now()) : null; + + if (batchOptions.enabled()) { + logger.info( + "Batching commits (schedule: {}, max files: {}, max bytes: {})", + batchOptions.commitSchedule() != null ? batchOptions.commitSchedule() : "unset", + batchOptions.maxFiles() > 0 ? String.valueOf(batchOptions.maxFiles()) : "unset", + batchOptions.maxBytes() > 0 + ? InsertWatchBuffer.formatBytes(batchOptions.maxBytes()) + : "unset"); + if (nextCommitAt != null) { + logger.info("Next commit scheduled for: {}", nextCommitAt); + } + } + final SqsClient sqs = buildSqsClient(sqsOverrideEndpoint); ReceiveMessageRequest req = ReceiveMessageRequest.builder() @@ -121,6 +158,8 @@ public static void run( // TODO: implement }; + final InsertWatchBuffer buffer = new InsertWatchBuffer(); + //noinspection LoopConditionNotUpdatedInsideLoop do { List batch = new LinkedList<>(); @@ -143,8 +182,9 @@ public static void run( Thread.sleep(delay); continue; } - if (!batch.isEmpty()) { - try { + + try { + if (!batch.isEmpty()) { var maxBatchSize = 100; // FIXME: make configurable List tailMessages; @@ -160,91 +200,201 @@ public static void run( logger.info("Processing {} message(s)", batch.size()); // FIXME: handle files not found - var insertBatch = filter(batch, matchers, metrics, tableLabel, queueLabel, queueType); - if (!insertBatch.isEmpty()) { - logger.info("Inserting {}", insertBatch); - - try { - Insert.Result result = - Insert.run(catalog, nsTable, insertBatch.toArray(String[]::new), options); - if (metrics != null) { - metrics.recordFilesInserted(tableLabel, queueLabel, queueType, insertBatch.size()); - metrics.recordTransactionSuccess(tableLabel, queueLabel, queueType); - } - if (!result.ok()) { - logger.warn( - "{}/{} file(s) failed to insert in this batch", - result.totalNumberOfFiles(), - result.numberOfFilesFailedToInsert()); - } - } catch (NoSuchTableException e) { - if (!createTableIfNotExists) { - if (metrics != null) { - metrics.recordTransactionFailed(tableLabel, queueLabel, queueType); - } - throw e; - } - boolean retryInsert = true; - try { - CreateTable.run( - catalog, - nsTable, - insertBatch.iterator().next(), - null, - true, - options.useVendedCredentials(), - options.s3NoSignRequest(), - null, - null); - } catch (NotFoundException nfe) { - if (!options.ignoreNotFound()) { - if (metrics != null) { - metrics.recordTransactionFailed(tableLabel, queueLabel, queueType); - } - throw nfe; - } - logger.info("Table not created ({} don't exist)", insertBatch); - retryInsert = false; - } - if (retryInsert) { - Insert.run(catalog, nsTable, insertBatch.toArray(String[]::new), options); - if (metrics != null) { - metrics.recordFilesInserted( - tableLabel, queueLabel, queueType, insertBatch.size()); - metrics.recordTransactionSuccess(tableLabel, queueLabel, queueType); - } - } - } + var filtered = filter(batch, matchers, metrics, tableLabel, queueLabel, queueType); + + // These contribute nothing to the next commit, so there is no reason to hold on to + // them until it happens. + confirmProcessed( + sqs, + sqsQueueURL, + filtered.unmatchedMessages(), + metrics, + tableLabel, + queueLabel, + queueType); + + buffer.add(filtered); + } + + boolean scheduleDue = nextCommitAt != null && !ZonedDateTime.now().isBefore(nextCommitAt); + String trigger = buffer.flushTrigger(batchOptions, scheduleDue); + if (trigger == null && terminateAfterOneBatch && !buffer.isEmpty()) { + trigger = "fire_once"; + } + if (trigger != null) { + flush( + catalog, + nsTable, + sqs, + sqsQueueURL, + buffer, + createTableIfNotExists, + options, + metrics, + tableLabel, + queueLabel, + queueType, + trigger); + if (schedule != null) { + nextCommitAt = rollForward(schedule, nextCommitAt, ZonedDateTime.now()); + logger.info("Next commit scheduled for: {}", nextCommitAt); } + } else if (!buffer.isEmpty()) { + keepInvisible(sqs, sqsQueueURL, buffer, nextCommitAt); + logBufferState(buffer, nextCommitAt); + } else if (scheduleDue && schedule != null) { + nextCommitAt = rollForward(schedule, nextCommitAt, ZonedDateTime.now()); + } + if (metrics != null) { + metrics.recordBufferState( + tableLabel, queueLabel, queueType, buffer.fileCount(), buffer.bytes()); + } + } catch (InterruptedException e) { + // terminate + Thread.currentThread().interrupt(); + throw new InterruptedException(); + } catch (Exception e) { + if (metrics != null) { + metrics.recordTransactionFailed(tableLabel, queueLabel, queueType); + metrics.recordRetryAttempt(tableLabel, queueLabel, queueType); + } + Duration delay = backoff.get(); + logger.error("Failed to process batch of messages (retry in {})", delay, e); + Thread.sleep(delay); + continue; + } + resetBackoff.run(); + } while (!terminateAfterOneBatch); + } + + /** + * Advances {@code deadline} from itself (not from {@code now}) so that a relative schedule such + * as {@code every 5 minutes} keeps a fixed cadence instead of sliding forward on every poll. + */ + private static ZonedDateTime rollForward(Schedule s, ZonedDateTime deadline, ZonedDateTime now) { + ZonedDateTime r = deadline; + while (!now.isBefore(r)) { + r = s.next(r); + } + return r; + } - confirmProcessed(sqs, sqsQueueURL, batch, metrics, tableLabel, queueLabel, queueType); - } catch (InterruptedException e) { - // terminate - Thread.currentThread().interrupt(); - throw new InterruptedException(); - } catch (Exception e) { + /** Commits everything accumulated so far as a single snapshot and acknowledges the messages. */ + private static void flush( + RESTCatalog catalog, + TableIdentifier nsTable, + SqsClient sqs, + String sqsQueueURL, + InsertWatchBuffer buffer, + boolean createTableIfNotExists, + Insert.Options options, + InsertWatchMetrics metrics, + String tableLabel, + String queueLabel, + String queueType, + String trigger) + throws IOException, InterruptedException { + String[] files = buffer.fileArray(); + logger.info( + "Committing {} file(s) ({}) accumulated over {}s (trigger: {})", + files.length, + InsertWatchBuffer.formatBytes(buffer.bytes()), + buffer.age().toSeconds(), + trigger); + logger.info("Inserting {}", Arrays.asList(files)); + + insert( + catalog, + nsTable, + files, + createTableIfNotExists, + options, + metrics, + tableLabel, + queueLabel, + queueType); + + confirmProcessed( + sqs, sqsQueueURL, buffer.messageList(), metrics, tableLabel, queueLabel, queueType); + + if (metrics != null) { + metrics.recordBufferFlush(tableLabel, queueLabel, queueType, trigger); + } + buffer.clear(); + } + + private static void insert( + RESTCatalog catalog, + TableIdentifier nsTable, + String[] files, + boolean createTableIfNotExists, + Insert.Options options, + InsertWatchMetrics metrics, + String tableLabel, + String queueLabel, + String queueType) + throws IOException, InterruptedException { + try { + Insert.Result result = Insert.run(catalog, nsTable, files, options); + if (metrics != null) { + metrics.recordFilesInserted(tableLabel, queueLabel, queueType, files.length); + metrics.recordTransactionSuccess(tableLabel, queueLabel, queueType); + } + if (!result.ok()) { + logger.warn( + "{}/{} file(s) failed to insert in this batch", + result.totalNumberOfFiles(), + result.numberOfFilesFailedToInsert()); + } + } catch (NoSuchTableException e) { + if (!createTableIfNotExists) { + if (metrics != null) { + metrics.recordTransactionFailed(tableLabel, queueLabel, queueType); + } + throw e; + } + boolean retryInsert = true; + try { + CreateTable.run( + catalog, + nsTable, + files[0], + null, + true, + options.useVendedCredentials(), + options.s3NoSignRequest(), + null, + null); + } catch (NotFoundException nfe) { + if (!options.ignoreNotFound()) { if (metrics != null) { metrics.recordTransactionFailed(tableLabel, queueLabel, queueType); - metrics.recordRetryAttempt(tableLabel, queueLabel, queueType); } - Duration delay = backoff.get(); - logger.error("Failed to process batch of messages (retry in {})", delay, e); - Thread.sleep(delay); - continue; + throw nfe; } + logger.info("Table not created ({} don't exist)", Arrays.asList(files)); + retryInsert = false; } - resetBackoff.run(); - } while (!terminateAfterOneBatch); + if (retryInsert) { + Insert.run(catalog, nsTable, files, options); + if (metrics != null) { + metrics.recordFilesInserted(tableLabel, queueLabel, queueType, files.length); + metrics.recordTransactionSuccess(tableLabel, queueLabel, queueType); + } + } + } } - private static Collection filter( + private static FilterResult filter( List messages, Collection matchers, InsertWatchMetrics metrics, String tableLabel, String queueLabel, String queueType) { - Collection r = new LinkedHashSet<>(); + Map files = new LinkedHashMap<>(); + List matched = new ArrayList<>(); + List unmatched = new ArrayList<>(); for (Message message : messages) { // Message body() example: // @@ -274,8 +424,10 @@ private static Collection filter( metrics.recordMessageParseError(tableLabel, queueLabel, queueType); } // TODO: dlq? + unmatched.add(message); continue; } + boolean messageMatched = false; // TODO: use type for (JsonNode record : root.path("Records")) { if (metrics != null) { @@ -291,7 +443,8 @@ private static Collection filter( if (eventName.startsWith("ObjectCreated:")) { // TODO: exclude metadata/data dirs by default if (matchers.stream().anyMatch(matcher -> matcher.test(target))) { - r.add(target); + files.putIfAbsent(target, record.at("/s3/object/size").asLong(0)); + messageMatched = true; if (metrics != null) { metrics.recordEventMatched(tableLabel, queueLabel, queueType); } @@ -310,8 +463,52 @@ private static Collection filter( } } } + (messageMatched ? matched : unmatched).add(message); } - return r; + return new FilterResult(files, matched, unmatched); + } + + /** + * Resets the visibility timeout of accumulated messages so that they are not redelivered while + * waiting for the next commit. + */ + private static void keepInvisible( + SqsClient sqs, String sqsQueueURL, InsertWatchBuffer buffer, ZonedDateTime nextCommitAt) { + int timeout = visibilityTimeoutSeconds(nextCommitAt); + List messages = buffer.messageList(); + int len = messages.size(); + for (int i = 0; i < len; i = i + SQS_BATCH_LIMIT) { + List chunk = messages.subList(i, Math.min(i + SQS_BATCH_LIMIT, len)); + // A message that stays visible is redelivered and re-accumulated rather than lost, so this + // is not worth failing the batch over. + try { + ChangeMessageVisibilityBatchResponse res = + changeMessageVisibilityBatch(sqs, sqsQueueURL, chunk, timeout); + for (BatchResultErrorEntry f : res.failed()) { + logger.warn("Failed to extend visibility of message#{}: {}", f.id(), f.message()); + } + } catch (SdkException e) { + logger.warn("Failed to extend visibility of {} accumulated message(s)", chunk.size(), e); + } + } + } + + private static int visibilityTimeoutSeconds(ZonedDateTime nextCommitAt) { + if (nextCommitAt == null) { + return MIN_VISIBILITY_TIMEOUT_SECONDS; + } + long secsUntil = Duration.between(ZonedDateTime.now(), nextCommitAt).toSeconds(); + long v = Math.max(MIN_VISIBILITY_TIMEOUT_SECONDS, secsUntil * 2); + return (int) Math.min(v, MAX_VISIBILITY_TIMEOUT_SECONDS); + } + + private static void logBufferState(InsertWatchBuffer buffer, ZonedDateTime nextCommitAt) { + logger.info( + "Accumulated {} file(s) ({}) over {}s; next commit at {}", + buffer.fileCount(), + InsertWatchBuffer.formatBytes(buffer.bytes()), + buffer.age().toSeconds(), + nextCommitAt != null ? nextCommitAt : "n/a"); } private static void confirmProcessed( @@ -322,10 +519,13 @@ private static void confirmProcessed( String tableLabel, String queueLabel, String queueType) { + if (messages.isEmpty()) { + return; + } int failedCount = 0; int len = messages.size(); - for (int i = 0; i < len; i = i + 10) { - List batch = messages.subList(i, Math.min(i + 10, len)); + for (int i = 0; i < len; i = i + SQS_BATCH_LIMIT) { + List batch = messages.subList(i, Math.min(i + SQS_BATCH_LIMIT, len)); DeleteMessageBatchResponse res = deleteMessageBatch(sqs, sqsQueueURL, batch); if (res.hasFailed()) { List failed = res.failed(); @@ -358,6 +558,24 @@ private static DeleteMessageBatchResponse deleteMessageBatch( .build()); } + private static ChangeMessageVisibilityBatchResponse changeMessageVisibilityBatch( + SqsClient sqs, String sqsQueueURL, List messages, int visibilityTimeoutSeconds) { + return sqs.changeMessageVisibilityBatch( + ChangeMessageVisibilityBatchRequest.builder() + .queueUrl(sqsQueueURL) + .entries( + messages.stream() + .map( + m -> + ChangeMessageVisibilityBatchRequestEntry.builder() + .id(m.messageId()) + .receiptHandle(m.receiptHandle()) + .visibilityTimeout(visibilityTimeoutSeconds) + .build()) + .toList()) + .build()); + } + private static SqsClient buildSqsClient(String sqsOverrideEndpoint) { SqsClientBuilder builder = SqsClient.builder(); diff --git a/ice/src/main/java/com/altinity/ice/cli/internal/cmd/InsertWatchBuffer.java b/ice/src/main/java/com/altinity/ice/cli/internal/cmd/InsertWatchBuffer.java new file mode 100644 index 00000000..b786e6a1 --- /dev/null +++ b/ice/src/main/java/com/altinity/ice/cli/internal/cmd/InsertWatchBuffer.java @@ -0,0 +1,151 @@ +/* + * Copyright (c) 2025 Altinity Inc and/or its affiliates. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + */ +package com.altinity.ice.cli.internal.cmd; + +import com.github.shyiko.skedule.Schedule; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import javax.annotation.Nullable; +import software.amazon.awssdk.services.sqs.model.Message; + +/** + * Files (and the messages that carried them) accumulated since the last Iceberg commit. + * + *

Keyed by message id so that a message redelivered before the commit does not end up twice in + * the delete request, which SQS rejects for having duplicate entry ids. + */ +public final class InsertWatchBuffer { + + /** + * Thresholds controlling how much data is accumulated before it is committed. Committing on every + * poll produces one Iceberg snapshot (and at least one manifest) per poll, which makes readers + * spend most of their time resolving metadata; accumulating first trades write latency for a + * proportionally smaller metadata tree. + */ + public record BatchOptions(@Nullable String commitSchedule, int maxFiles, long maxBytes) { + + public static final BatchOptions NONE = new BatchOptions(null, 0, 0); + + public BatchOptions { + if (maxFiles < 0) { + throw new IllegalArgumentException("--watch-max-files must be non-negative"); + } + if (maxBytes < 0) { + throw new IllegalArgumentException("--watch-max-bytes must be non-negative"); + } + if (commitSchedule != null) { + // Fail fast on a bad expression at startup rather than on the first poll. + Schedule.parse(commitSchedule); + } + } + + public boolean enabled() { + return commitSchedule != null || maxFiles > 0 || maxBytes > 0; + } + } + + /** + * Result of matching a poll batch against the input patterns: the files to insert (mapped to the + * size reported by the S3 event), the messages that carried them, and the messages that can be + * acknowledged immediately because nothing in them matched. + */ + record FilterResult( + Map files, List messages, List unmatchedMessages) {} + + private final Map files = new LinkedHashMap<>(); + private final Map messages = new LinkedHashMap<>(); + private long bytes; + private Instant startedAt; + + void add(FilterResult r) { + for (var e : r.files().entrySet()) { + if (files.putIfAbsent(e.getKey(), e.getValue()) == null) { + bytes += e.getValue(); + } + } + for (Message m : r.messages()) { + messages.put(m.messageId(), m); + } + if (startedAt == null && !files.isEmpty()) { + startedAt = Instant.now(); + } + } + + boolean isEmpty() { + return files.isEmpty(); + } + + int fileCount() { + return files.size(); + } + + long bytes() { + return bytes; + } + + Duration age() { + return startedAt == null ? Duration.ZERO : Duration.between(startedAt, Instant.now()); + } + + String[] fileArray() { + return files.keySet().toArray(String[]::new); + } + + List messageList() { + return new ArrayList<>(messages.values()); + } + + void clear() { + files.clear(); + messages.clear(); + bytes = 0; + startedAt = null; + } + + /** Returns the threshold that was reached, or null if the buffer should keep filling up. */ + @Nullable + String flushTrigger(BatchOptions o, boolean scheduleDue) { + if (files.isEmpty()) { + return null; + } + if (!o.enabled()) { + return "immediate"; + } + if (o.maxFiles() > 0 && files.size() >= o.maxFiles()) { + return "max_files"; + } + if (o.maxBytes() > 0 && bytes >= o.maxBytes()) { + return "max_bytes"; + } + if (scheduleDue) { + return "schedule"; + } + return null; + } + + static String formatBytes(long bytes) { + if (bytes < 1024) { + return bytes + " B"; + } + String[] units = {"KiB", "MiB", "GiB", "TiB", "PiB"}; + double v = bytes; + int i = -1; + while (v >= 1024 && i < units.length - 1) { + v /= 1024; + i++; + } + return String.format(Locale.ENGLISH, "%.1f %s", v, units[i]); + } +} diff --git a/ice/src/main/java/com/altinity/ice/cli/internal/metrics/InsertWatchMetrics.java b/ice/src/main/java/com/altinity/ice/cli/internal/metrics/InsertWatchMetrics.java index 483bc465..f301fad6 100644 --- a/ice/src/main/java/com/altinity/ice/cli/internal/metrics/InsertWatchMetrics.java +++ b/ice/src/main/java/com/altinity/ice/cli/internal/metrics/InsertWatchMetrics.java @@ -10,6 +10,7 @@ package com.altinity.ice.cli.internal.metrics; import io.prometheus.metrics.core.metrics.Counter; +import io.prometheus.metrics.core.metrics.Gauge; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -34,9 +35,14 @@ private static class Holder { private static final String LABEL_TABLE = "table"; private static final String LABEL_QUEUE = "queue"; private static final String LABEL_QUEUE_TYPE = "queue_type"; + private static final String LABEL_TRIGGER = "trigger"; private static final String[] WATCH_LABELS = {LABEL_TABLE, LABEL_QUEUE, LABEL_QUEUE_TYPE}; + private static final String[] FLUSH_LABELS = { + LABEL_TABLE, LABEL_QUEUE, LABEL_QUEUE_TYPE, LABEL_TRIGGER + }; + // Messages/Files processed private static final String MESSAGES_RECEIVED_TOTAL_NAME = "ice_watch_messages_received_total"; private static final String MESSAGES_RECEIVED_TOTAL_HELP = @@ -73,6 +79,19 @@ private static class Holder { private static final String TRANSACTIONS_FAILED_TOTAL_HELP = "Total number of insert transactions that failed"; + // Commit batching + private static final String BUFFER_FILES_NAME = "ice_watch_buffer_files"; + private static final String BUFFER_FILES_HELP = + "Number of files accumulated and waiting to be committed"; + + private static final String BUFFER_BYTES_NAME = "ice_watch_buffer_bytes"; + private static final String BUFFER_BYTES_HELP = + "Total size of the files accumulated and waiting to be committed"; + + private static final String BUFFER_FLUSHES_TOTAL_NAME = "ice_watch_buffer_flushes_total"; + private static final String BUFFER_FLUSHES_TOTAL_HELP = + "Total number of accumulated batches committed, by the threshold that triggered the commit"; + // Retry state private static final String RETRY_ATTEMPTS_TOTAL_NAME = "ice_watch_retry_attempts_total"; private static final String RETRY_ATTEMPTS_TOTAL_HELP = @@ -112,6 +131,9 @@ private static class Holder { private final Counter filesInsertedTotal; private final Counter transactionsTotal; private final Counter transactionsFailedTotal; + private final Gauge bufferFiles; + private final Gauge bufferBytes; + private final Counter bufferFlushesTotal; private final Counter retryAttemptsTotal; private final Counter queueReceiveErrorsTotal; private final Counter queueDeleteErrorsTotal; @@ -180,6 +202,27 @@ private InsertWatchMetrics() { .labelNames(WATCH_LABELS) .register(); + this.bufferFiles = + Gauge.builder() + .name(BUFFER_FILES_NAME) + .help(BUFFER_FILES_HELP) + .labelNames(WATCH_LABELS) + .register(); + + this.bufferBytes = + Gauge.builder() + .name(BUFFER_BYTES_NAME) + .help(BUFFER_BYTES_HELP) + .labelNames(WATCH_LABELS) + .register(); + + this.bufferFlushesTotal = + Counter.builder() + .name(BUFFER_FLUSHES_TOTAL_NAME) + .help(BUFFER_FLUSHES_TOTAL_HELP) + .labelNames(FLUSH_LABELS) + .register(); + this.retryAttemptsTotal = Counter.builder() .name(RETRY_ATTEMPTS_TOTAL_NAME) @@ -250,6 +293,16 @@ public void recordTransactionFailed(String table, String queue, String queueType transactionsFailedTotal.labelValues(table, queue, queueType).inc(); } + public void recordBufferState( + String table, String queue, String queueType, int files, long bytes) { + bufferFiles.labelValues(table, queue, queueType).set(files); + bufferBytes.labelValues(table, queue, queueType).set(bytes); + } + + public void recordBufferFlush(String table, String queue, String queueType, String trigger) { + bufferFlushesTotal.labelValues(table, queue, queueType, trigger).inc(); + } + public void recordRetryAttempt(String table, String queue, String queueType) { retryAttemptsTotal.labelValues(table, queue, queueType).inc(); } diff --git a/pom.xml b/pom.xml index be0c3dd8..ff8464e4 100644 --- a/pom.xml +++ b/pom.xml @@ -26,6 +26,7 @@ 11.0.25 6.1.0 2.60.0 + 0.4.0 7.9.0 3.27.7 From 1b52cafe674140980bbd77133948a7165e9c63b4 Mon Sep 17 00:00:00 2001 From: kanthi subramanian Date: Tue, 18 Aug 2026 12:24:52 -0500 Subject: [PATCH 2/5] Fix integration test of DockerElasticMQWatchIT --- .github/workflows/verify.yaml | 2 +- ice-rest-catalog/pom.xml | 1 + .../rest/catalog/DockerElasticMQWatchIT.java | 369 ++++++++++++++++++ .../src/test/resources/elasticmq.conf | 43 ++ 4 files changed, 414 insertions(+), 1 deletion(-) create mode 100644 ice-rest-catalog/src/test/java/com/altinity/ice/rest/catalog/DockerElasticMQWatchIT.java create mode 100644 ice-rest-catalog/src/test/resources/elasticmq.conf diff --git a/.github/workflows/verify.yaml b/.github/workflows/verify.yaml index 0c31f2a5..0995f870 100644 --- a/.github/workflows/verify.yaml +++ b/.github/workflows/verify.yaml @@ -55,6 +55,6 @@ jobs: run: > ./mvnw -pl ice-rest-catalog -am install -DskipTests=true -Pno-check && ./mvnw -pl ice-rest-catalog failsafe:integration-test failsafe:verify - -Dit.test=DockerScenarioBasedIT,DockerLocalFileIOClickHouseIT,DockerLocalFileIOClickHouseAllTypesIT + -Dit.test=DockerScenarioBasedIT,DockerLocalFileIOClickHouseIT,DockerLocalFileIOClickHouseAllTypesIT,DockerElasticMQWatchIT -Ddocker.image=altinity/ice-rest-catalog:debug-with-ice-latest-master-amd64 -Dclickhouse.image=altinity/clickhouse-server:25.8.16.20002.altinityantalya diff --git a/ice-rest-catalog/pom.xml b/ice-rest-catalog/pom.xml index 66288b7a..aa42b90f 100644 --- a/ice-rest-catalog/pom.xml +++ b/ice-rest-catalog/pom.xml @@ -598,6 +598,7 @@ **/DockerScenarioBasedIT.java **/DockerLocalFileIOClickHouseIT.java **/DockerLocalFileIOClickHouseAllTypesIT.java + **/DockerElasticMQWatchIT.java diff --git a/ice-rest-catalog/src/test/java/com/altinity/ice/rest/catalog/DockerElasticMQWatchIT.java b/ice-rest-catalog/src/test/java/com/altinity/ice/rest/catalog/DockerElasticMQWatchIT.java new file mode 100644 index 00000000..078a4716 --- /dev/null +++ b/ice-rest-catalog/src/test/java/com/altinity/ice/rest/catalog/DockerElasticMQWatchIT.java @@ -0,0 +1,369 @@ +/* + * Copyright (c) 2025 Altinity Inc and/or its affiliates. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + */ +package com.altinity.ice.rest.catalog; + +import java.io.IOException; +import java.net.URI; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.MetricsConfig; +import org.apache.iceberg.Schema; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.data.parquet.GenericParquetWriter; +import org.apache.iceberg.hadoop.HadoopOutputFile; +import org.apache.iceberg.io.FileAppender; +import org.apache.iceberg.parquet.Parquet; +import org.apache.iceberg.types.Types; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.Container.ExecResult; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.Network; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.utility.MountableFile; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.core.sync.RequestBody; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.model.CreateBucketRequest; +import software.amazon.awssdk.services.s3.model.PutObjectRequest; +import software.amazon.awssdk.services.sqs.SqsClient; +import software.amazon.awssdk.services.sqs.model.SendMessageRequest; + +/** + * Docker integration test: {@code ice insert --watch} consuming an S3 object-created event from an + * SQS-compatible queue (ElasticMQ) and committing the referenced Parquet file as an Iceberg + * snapshot. + * + *

Topology: MinIO holds the warehouse ({@code s3://test-bucket/warehouse}) and the landing + * object ({@code s3://test-bucket/landing/data.parquet}); ElasticMQ provides the {@code s3-events} + * queue; the catalog container runs the co-located {@code ice} CLI in watch mode against ElasticMQ. + * The watch is driven with {@code --watch-commit-schedule} + {@code --watch-max-files=1} so the + * first matched file flushes immediately, and {@code --watch-fire-once} so the command exits after + * one poll cycle instead of looping forever. + * + *

The watch batching flags are client-side in the {@code ice} CLI, so the catalog image must be + * built from current source for the bundled {@code ice} to have them. Build it locally once (per + * {@code ice} change) with: + * + * 

docker build --build-arg BASE_IMAGE_TAG=debug \
+ *   -t altinity/ice-rest-catalog:debug-with-ice-local \
+ *   -f ice-rest-catalog/Dockerfile.debug-with-ice .
+ * + *

Requires Docker. Excluded from default Failsafe runs (see {@code pom.xml}); run explicitly, + * e.g. {@code mvn -pl ice-rest-catalog failsafe:integration-test failsafe:verify + * -Dit.test=DockerElasticMQWatchIT}. Image tags can be overridden via {@code -Ddocker.image=...}, + * {@code -Delasticmq.image=...} and {@code -Dminio.image=...}. + */ +public class DockerElasticMQWatchIT { + + private static final Logger logger = LoggerFactory.getLogger(DockerElasticMQWatchIT.class); + + private static final String DEFAULT_CATALOG_IMAGE = + "altinity/ice-rest-catalog:debug-with-ice-local"; + private static final String DEFAULT_ELASTICMQ_IMAGE = "softwaremill/elasticmq-native:1.6.15"; + private static final String DEFAULT_MINIO_IMAGE = "minio/minio:latest"; + + private static final String BUCKET = "test-bucket"; + private static final String QUEUE_NAME = "s3-events"; + private static final String NAMESPACE = "watch_test"; + private static final String TABLE = NAMESPACE + ".events"; + private static final String LANDING_KEY = "warehouse/watch_test/events/external/data.parquet"; + + // ElasticMQ queue URL as seen from inside the Docker network (see elasticmq.conf accountId). + private static final String QUEUE_URL_INTERNAL = + "http://elasticmq:9324/000000000000/" + QUEUE_NAME; + + private Network network; + private GenericContainer minio; + private GenericContainer elasticmq; + private GenericContainer catalog; + + @BeforeClass + @SuppressWarnings("resource") + public void setUp() throws Exception { + String dockerImage = System.getProperty("docker.image", DEFAULT_CATALOG_IMAGE); + String elasticmqImage = System.getProperty("elasticmq.image", DEFAULT_ELASTICMQ_IMAGE); + String minioImage = System.getProperty("minio.image", DEFAULT_MINIO_IMAGE); + logger.info( + "Using images: catalog={}, elasticmq={}, minio={}", + dockerImage, + elasticmqImage, + minioImage); + + network = Network.newNetwork(); + + minio = + new GenericContainer<>(minioImage) + .withNetwork(network) + .withNetworkAliases("minio") + .withExposedPorts(9000) + .withEnv("MINIO_ACCESS_KEY", "minioadmin") + .withEnv("MINIO_SECRET_KEY", "minioadmin") + .withCommand("server", "/data") + .waitingFor(Wait.forHttp("/minio/health/live").forPort(9000)); + minio.start(); + + String minioHostEndpoint = "http://" + minio.getHost() + ":" + minio.getMappedPort(9000); + try (S3Client s3 = minioS3(minioHostEndpoint)) { + s3.createBucket(CreateBucketRequest.builder().bucket(BUCKET).build()); + logger.info("Created bucket {} in MinIO", BUCKET); + } + + elasticmq = + new GenericContainer<>(elasticmqImage) + .withNetwork(network) + .withNetworkAliases("elasticmq") + .withExposedPorts(9324) + .withCopyFileToContainer( + MountableFile.forClasspathResource("elasticmq.conf"), "/opt/elasticmq.conf") + .withCommand("-Dconfig.file=/opt/elasticmq.conf") + .waitingFor(Wait.forListeningPort()); + elasticmq.start(); + + URL configResource = getClass().getClassLoader().getResource("docker-catalog-config.yaml"); + if (configResource == null) { + throw new IllegalStateException("docker-catalog-config.yaml not on classpath"); + } + String catalogConfig = Files.readString(Paths.get(configResource.toURI())); + + catalog = + new GenericContainer<>(dockerImage) + .withNetwork(network) + .withNetworkAliases("catalog") + .withExposedPorts(5000) + .withEnv("ICE_REST_CATALOG_CONFIG", "") + .withEnv("ICE_REST_CATALOG_CONFIG_YAML", catalogConfig) + // Default AWS credential chain used by the watcher's SQS client. Values match MinIO so + // the same env also satisfies any incidental S3 use; ElasticMQ accepts any well-formed + // credentials. + .withEnv("AWS_ACCESS_KEY_ID", "minioadmin") + .withEnv("AWS_SECRET_ACCESS_KEY", "minioadmin") + .withEnv("AWS_REGION", "us-east-1") + .waitingFor(Wait.forHttp("/v1/config").forPort(5000).forStatusCode(200)); + + try { + catalog.start(); + } catch (Exception e) { + logger.error("Catalog container logs: {}", catalog.getLogs()); + throw e; + } + + // CLI config: ice runs inside the catalog container, so it reaches MinIO and ElasticMQ via + // their network aliases. + String cliConfig = + "uri: http://localhost:5000\n" + + "warehouse: s3://" + + BUCKET + + "/warehouse\n" + + "s3:\n" + + " endpoint: http://minio:9000\n" + + " pathStyleAccess: true\n" + + " accessKeyID: minioadmin\n" + + " secretAccessKey: minioadmin\n" + + " region: us-east-1\n"; + catalog.copyFileToContainer( + MountableFile.forHostPath(writeTemp(cliConfig)), "/tmp/ice-cli.yaml"); + + logger.info( + "Catalog at {}:{}, ElasticMQ at {}:{}", + catalog.getHost(), + catalog.getMappedPort(5000), + elasticmq.getHost(), + elasticmq.getMappedPort(9324)); + } + + @AfterClass + public void tearDown() { + if (catalog != null) { + catalog.close(); + } + if (elasticmq != null) { + elasticmq.close(); + } + if (minio != null) { + minio.close(); + } + if (network != null) { + network.close(); + } + } + + @Test + public void testWatchCommitsS3EventAsSnapshot() throws Exception { + Path parquet = Files.createTempFile("watch-it-", ".parquet"); + try { + writeParquet(parquet); + long size = Files.size(parquet); + + // Upload the file to MinIO under the warehouse prefix so the noCopy insert path uses + // table.io() (REST-catalog-configured FileIO with the MinIO endpoint) rather than a raw + // S3FileIO that lacks the endpoint override. + String minioHostEndpoint = "http://" + minio.getHost() + ":" + minio.getMappedPort(9000); + try (S3Client s3 = minioS3(minioHostEndpoint)) { + s3.putObject( + PutObjectRequest.builder().bucket(BUCKET).key(LANDING_KEY).build(), + RequestBody.fromFile(parquet)); + } + logger.info("Uploaded {} to s3://{}/{} ({} bytes)", parquet, BUCKET, LANDING_KEY, size); + + // Pre-create the table from a local copy so CreateTable.run (which builds its own S3Client + // without the MinIO endpoint) is never invoked during the watch flush. + catalog.copyFileToContainer(MountableFile.forHostPath(parquet), "/tmp/seed.parquet"); + iceExecOrThrow("create-namespace", NAMESPACE); + iceExecOrThrow("insert", "--create-table", TABLE, "file:///tmp/seed.parquet"); + + // Notify the watcher about the new object. + String elasticmqHostEndpoint = + "http://" + elasticmq.getHost() + ":" + elasticmq.getMappedPort(9324); + String queueUrlHost = elasticmqHostEndpoint + "/000000000000/" + QUEUE_NAME; + try (SqsClient sqs = + SqsClient.builder() + .endpointOverride(URI.create(elasticmqHostEndpoint)) + .region(Region.US_EAST_1) + .credentialsProvider( + StaticCredentialsProvider.create( + AwsBasicCredentials.create("minioadmin", "minioadmin"))) + .build()) { + sqs.sendMessage( + SendMessageRequest.builder() + .queueUrl(queueUrlHost) + .messageBody(s3Event(BUCKET, LANDING_KEY, size)) + .build()); + } + logger.info("Sent S3 event for s3://{}/{} to {}", BUCKET, LANDING_KEY, queueUrlHost); + + // Run the watcher once. --watch-max-files=1 flushes the first matched file immediately; + // --watch-fire-once exits after the first poll cycle. + ExecResult watch = + ice( + "insert", + TABLE, + "-p", + "--force-no-copy", + "--skip-duplicates", + "--watch=" + QUEUE_URL_INTERNAL, + "--watch-endpoint=http://elasticmq:9324", + "--watch-commit-schedule=every 5 minutes", + "--watch-max-files=1", + "--watch-fire-once", + "s3://" + BUCKET + "/warehouse/watch_test/events/external/*.parquet"); + logger.info("watch stdout:\n{}", watch.getStdout()); + logger.info("watch stderr:\n{}", watch.getStderr()); + if (watch.getExitCode() != 0) { + throw new AssertionError( + "ice insert --watch exited " + watch.getExitCode() + ":\n" + watch.getStderr()); + } + + ExecResult scan = iceExecOrThrow("scan", TABLE); + logger.info("scan stdout:\n{}", scan.getStdout()); + if (!scan.getStdout().contains("watch-it")) { + throw new AssertionError( + "Expected committed row in scan output, got:\n" + scan.getStdout()); + } + } finally { + Files.deleteIfExists(parquet); + } + } + + private static String s3Event(String bucket, String key, long size) { + return "{\"Records\":[{\"eventName\":\"ObjectCreated:Put\"," + + "\"eventTime\":\"2026-08-16T00:00:00.000Z\"," + + "\"s3\":{\"bucket\":{\"name\":\"" + + bucket + + "\"},\"object\":{\"key\":\"" + + key + + "\",\"size\":" + + size + + "}}}]}"; + } + + private static S3Client minioS3(String endpoint) { + return S3Client.builder() + .endpointOverride(URI.create(endpoint)) + .region(Region.US_EAST_1) + .credentialsProvider( + StaticCredentialsProvider.create( + AwsBasicCredentials.create("minioadmin", "minioadmin"))) + .forcePathStyle(true) + .build(); + } + + /** Runs the catalog container's bundled {@code ice} CLI against the test config. */ + private ExecResult ice(String... args) throws IOException, InterruptedException { + List cmd = new ArrayList<>(); + cmd.add("ice"); + cmd.add("--config"); + cmd.add("/tmp/ice-cli.yaml"); + for (String a : args) { + cmd.add(a); + } + return catalog.execInContainer(cmd.toArray(new String[0])); + } + + private ExecResult iceExecOrThrow(String... args) throws IOException, InterruptedException { + ExecResult r = ice(args); + logger.info("ice {} stdout:\n{}", String.join(" ", args), r.getStdout()); + if (r.getExitCode() != 0) { + throw new IllegalStateException( + "ice " + + String.join(" ", args) + + " failed: exit=" + + r.getExitCode() + + "\nstdout:\n" + + r.getStdout() + + "\nstderr:\n" + + r.getStderr() + + "\ncatalog logs:\n" + + catalog.getLogs()); + } + return r; + } + + private static Path writeTemp(String contents) throws IOException { + Path f = Files.createTempFile("ice-watch-cli-", ".yaml"); + Files.writeString(f, contents); + f.toFile().deleteOnExit(); + return f; + } + + private static void writeParquet(Path file) throws IOException { + Schema schema = + new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.required(2, "name", Types.StringType.get())); + Record row = GenericRecord.create(schema); + row.setField("id", 1); + row.setField("name", "watch-it"); + org.apache.iceberg.io.OutputFile outputFile = + HadoopOutputFile.fromPath(new org.apache.hadoop.fs.Path(file.toUri()), new Configuration()); + try (FileAppender writer = + Parquet.write(outputFile) + .schema(schema) + .setAll(java.util.Map.of()) + .createWriterFunc(GenericParquetWriter::buildWriter) + .metricsConfig(MetricsConfig.getDefault()) + // Files.createTempFile already created the file, so allow overwriting it. + .overwrite() + .build()) { + writer.add(row); + } + } +} diff --git a/ice-rest-catalog/src/test/resources/elasticmq.conf b/ice-rest-catalog/src/test/resources/elasticmq.conf new file mode 100644 index 00000000..775202e1 --- /dev/null +++ b/ice-rest-catalog/src/test/resources/elasticmq.conf @@ -0,0 +1,43 @@ +include classpath("application.conf") + +node-address { + protocol = http + host = "*" + port = 9324 + context-path = "" +} + +rest-sqs { + enabled = true + bind-port = 9324 + bind-hostname = "0.0.0.0" + sqs-limits = strict +} + +rest-stats { + enabled = true + bind-port = 9325 + bind-hostname = "0.0.0.0" +} + +queues { + s3-events { + defaultVisibilityTimeout = 30 seconds + delay = 0 seconds + receiveMessageWait = 0 seconds + deadLettersQueue { + name = "s3-events-dlq" + maxReceiveCount = 3 + } + } + s3-events-dlq { + defaultVisibilityTimeout = 30 seconds + delay = 0 seconds + receiveMessageWait = 0 seconds + } +} + +aws { + region = elasticmq + accountId = 000000000000 +} From c8980be5946658dd17cf51bb2f69bef2deb62439 Mon Sep 17 00:00:00 2001 From: kanthi subramanian Date: Tue, 18 Aug 2026 12:34:21 -0500 Subject: [PATCH 3/5] Updated documentation on ice insert watch --- docs/insert-watch.md | 45 ++++++++++++++++++++++++++++++++++++++++++++ ice/README.md | 2 ++ 2 files changed, 47 insertions(+) create mode 100644 docs/insert-watch.md diff --git a/docs/insert-watch.md b/docs/insert-watch.md new file mode 100644 index 00000000..02321afd --- /dev/null +++ b/docs/insert-watch.md @@ -0,0 +1,45 @@ +# How `ice insert --watch` Works + +`ice insert --watch` long-polls an SQS queue for S3 object-create events and appends matching Parquet files to an Iceberg table **by reference** (no data copy). + +Implemented in [`InsertWatch.java`](../ice/src/main/java/com/altinity/ice/cli/internal/cmd/InsertWatch.java). AWS example: [`examples/s3watch`](../examples/s3watch/README.md). Local ElasticMQ: [`examples/s3watch/test`](../examples/s3watch/test/README.md). + +## Usage + +Requires `--no-copy` (register existing S3 objects) and `--skip-duplicates` (re-delivered messages must not fail). File arguments are **match patterns**, not a one-time list. + +```shell +ice insert flowers.iris -p --no-copy --skip-duplicates \ + s3://$BUCKET/flowers/iris/external-data/*.parquet \ + --watch="$SQS_QUEUE_URL" +``` + +| Flag | Purpose | +|------|---------| +| `--watch=` | SQS queue URL. | +| `--watch-endpoint=` | Custom SQS endpoint (ElasticMQ, LocalStack). | +| `--watch-fire-once` | One poll cycle, then exit. | +| `--watch-debug-addr=` | `/metrics`, `/healtz`, `/livez`, `/readyz`. Enables Prometheus metrics. | +| `--watch-commit-schedule=` | Accumulate files and commit them as one snapshot on this [skedule](https://github.com/shyiko/skedule) schedule, e.g. `"every 5 minutes"`, `"every day 02:00"`. Default: commit on every poll. | +| `--watch-max-files=` | Commit as soon as this many files are accumulated (default: no limit). | +| `--watch-max-bytes=` | Commit as soon as accumulated files add up to this many bytes (default: no limit). | +| `-p` | Create the table from the first matched file if it does not exist. | + +Watch mode also sets `ignoreNotFound=true` so a deleted object does not fail the batch. + +## Flow + +1. **Poll.** Long-poll SQS (`waitTime=20s`, max 10 messages), then short-poll drain until empty or 100 messages. +2. **Filter.** Parse each body as a raw S3 event (`Records[]`, not SNS-wrapped). Keep `ObjectCreated:*` events whose `s3://bucket/key` matches an input glob. Same key in one drain is inserted once. +3. **Accumulate / insert.** Matched files are held until a flush trigger fires (`--watch-commit-schedule`, `--watch-max-files`, `--watch-max-bytes`, or immediately when none of those are set). Then call the same `Insert.run` as a one-shot insert: read Parquet footers, skip paths already in the snapshot, append `DataFile`s pointing at the existing URIs, commit one snapshot for the whole accumulated batch. With `-p`, table create is delayed until the first matching file arrives. +4. **Ack.** Delete unmatched SQS messages immediately. Delete matched messages only after the accumulated batch commits. On insert/receive failure, do not delete; sleep 20s and retry after visibility timeout. + +`--no-copy` requires objects under `table.location()` unless `--force-no-copy` is set. + +## Notes + +- **At-least-once.** A crash after commit but before delete re-delivers the same keys; `--skip-duplicates` makes that a no-op. +- **Direct S3 → SQS only.** An SNS envelope has no top-level `Records`; those events are ignored. +- **No metadata/data exclusion.** Point the glob at the landing prefix, not the table warehouse root. +- **Visibility timeout.** While files are accumulated, the watcher extends SQS visibility for the held messages (at least 60s, doubled remaining time until the next scheduled commit, capped at 12h). Size the queue default larger than a worst-case insert of a single poll batch. +- Metrics (when `--watch-debug-addr` is set) are listed in [`examples/grafana/METRICS.md`](../examples/grafana/METRICS.md). diff --git a/ice/README.md b/ice/README.md index 8e0bfe16..f3e98c11 100644 --- a/ice/README.md +++ b/ice/README.md @@ -18,6 +18,8 @@ A CLI for loading data into Iceberg REST catalogs. - [Inspect](#inspect) - [S3 with Public Data](#s3-with-public-data) - [Describe Metadata](#describe-metadata) + - [Insert Watch](../docs/insert-watch.md) - Read messages from SQS and insert data into Iceberg tables. + ## Usage From 3019fcb97b7567e344727963f24f50eed5dd2330 Mon Sep 17 00:00:00 2001 From: kanthi subramanian Date: Tue, 18 Aug 2026 14:49:59 -0500 Subject: [PATCH 4/5] added test case for max bytes --- .../rest/catalog/DockerElasticMQWatchIT.java | 119 +++++++++++++++++- 1 file changed, 118 insertions(+), 1 deletion(-) diff --git a/ice-rest-catalog/src/test/java/com/altinity/ice/rest/catalog/DockerElasticMQWatchIT.java b/ice-rest-catalog/src/test/java/com/altinity/ice/rest/catalog/DockerElasticMQWatchIT.java index 078a4716..eb4a03cf 100644 --- a/ice-rest-catalog/src/test/java/com/altinity/ice/rest/catalog/DockerElasticMQWatchIT.java +++ b/ice-rest-catalog/src/test/java/com/altinity/ice/rest/catalog/DockerElasticMQWatchIT.java @@ -25,6 +25,7 @@ import org.apache.iceberg.data.parquet.GenericParquetWriter; import org.apache.iceberg.hadoop.HadoopOutputFile; import org.apache.iceberg.io.FileAppender; +import org.apache.iceberg.io.OutputFile; import org.apache.iceberg.parquet.Parquet; import org.apache.iceberg.types.Types; import org.slf4j.Logger; @@ -283,6 +284,122 @@ public void testWatchCommitsS3EventAsSnapshot() throws Exception { } } + @Test + public void testWatchMaxBytesTrigger() throws Exception { + String table = NAMESPACE + ".events_bytes"; + String landingKey = "warehouse/watch_test/events_bytes/external/data.parquet"; + long size = seedTableAndEnqueueEvent(table, landingKey); + + ExecResult watch = + ice( + "insert", + table, + "-p", + "--force-no-copy", + "--skip-duplicates", + "--watch=" + QUEUE_URL_INTERNAL, + "--watch-endpoint=http://elasticmq:9324", + "--watch-max-bytes=" + size, + "--watch-fire-once", + "s3://" + BUCKET + "/warehouse/watch_test/events_bytes/external/*.parquet"); + logger.info("watch stdout:\n{}", watch.getStdout()); + logger.info("watch stderr:\n{}", watch.getStderr()); + if (watch.getExitCode() != 0) { + throw new AssertionError( + "ice insert --watch exited " + watch.getExitCode() + ":\n" + watch.getStderr()); + } + if (!watch.getStderr().contains("trigger: max_bytes")) { + throw new AssertionError( + "Expected flush trigger 'max_bytes' in watch stderr, got:\n" + watch.getStderr()); + } + + ExecResult scan = iceExecOrThrow("scan", table); + if (!scan.getStdout().contains("watch-it")) { + throw new AssertionError("Expected committed row in scan output, got:\n" + scan.getStdout()); + } + } + + @Test + public void testWatchScheduleOnlyTrigger() throws Exception { + String table = NAMESPACE + ".events_sched"; + String landingKey = "warehouse/watch_test/events_sched/external/data.parquet"; + seedTableAndEnqueueEvent(table, landingKey); + + ExecResult watch = + ice( + "insert", + table, + "-p", + "--force-no-copy", + "--skip-duplicates", + "--watch=" + QUEUE_URL_INTERNAL, + "--watch-endpoint=http://elasticmq:9324", + "--watch-commit-schedule=every 1 minutes", + "--watch-fire-once", + "s3://" + BUCKET + "/warehouse/watch_test/events_sched/external/*.parquet"); + logger.info("watch stdout:\n{}", watch.getStdout()); + logger.info("watch stderr:\n{}", watch.getStderr()); + if (watch.getExitCode() != 0) { + throw new AssertionError( + "ice insert --watch exited " + watch.getExitCode() + ":\n" + watch.getStderr()); + } + if (!watch.getStderr().contains("trigger: fire_once")) { + throw new AssertionError( + "Expected flush trigger 'fire_once' in watch stderr, got:\n" + watch.getStderr()); + } + + ExecResult scan = iceExecOrThrow("scan", table); + if (!scan.getStdout().contains("watch-it")) { + throw new AssertionError("Expected committed row in scan output, got:\n" + scan.getStdout()); + } + } + + /** + * Writes a Parquet file, uploads it to MinIO, pre-creates the table from a local copy, and sends + * an S3 event to ElasticMQ. Returns the file size in bytes (useful for --watch-max-bytes). + */ + private long seedTableAndEnqueueEvent(String table, String landingKey) throws Exception { + Path parquet = Files.createTempFile("watch-it-", ".parquet"); + try { + writeParquet(parquet); + long size = Files.size(parquet); + + String minioHostEndpoint = "http://" + minio.getHost() + ":" + minio.getMappedPort(9000); + try (S3Client s3 = minioS3(minioHostEndpoint)) { + s3.putObject( + PutObjectRequest.builder().bucket(BUCKET).key(landingKey).build(), + RequestBody.fromFile(parquet)); + } + logger.info("Uploaded to s3://{}/{} ({} bytes)", BUCKET, landingKey, size); + + catalog.copyFileToContainer(MountableFile.forHostPath(parquet), "/tmp/seed.parquet"); + iceExecOrThrow("insert", "--create-table", table, "file:///tmp/seed.parquet"); + + String elasticmqHostEndpoint = + "http://" + elasticmq.getHost() + ":" + elasticmq.getMappedPort(9324); + String queueUrlHost = elasticmqHostEndpoint + "/000000000000/" + QUEUE_NAME; + try (SqsClient sqs = + SqsClient.builder() + .endpointOverride(URI.create(elasticmqHostEndpoint)) + .region(Region.US_EAST_1) + .credentialsProvider( + StaticCredentialsProvider.create( + AwsBasicCredentials.create("minioadmin", "minioadmin"))) + .build()) { + sqs.sendMessage( + SendMessageRequest.builder() + .queueUrl(queueUrlHost) + .messageBody(s3Event(BUCKET, landingKey, size)) + .build()); + } + logger.info("Sent S3 event for s3://{}/{}", BUCKET, landingKey); + + return size; + } finally { + Files.deleteIfExists(parquet); + } + } + private static String s3Event(String bucket, String key, long size) { return "{\"Records\":[{\"eventName\":\"ObjectCreated:Put\"," + "\"eventTime\":\"2026-08-16T00:00:00.000Z\"," @@ -352,7 +469,7 @@ private static void writeParquet(Path file) throws IOException { Record row = GenericRecord.create(schema); row.setField("id", 1); row.setField("name", "watch-it"); - org.apache.iceberg.io.OutputFile outputFile = + OutputFile outputFile = HadoopOutputFile.fromPath(new org.apache.hadoop.fs.Path(file.toUri()), new Configuration()); try (FileAppender writer = Parquet.write(outputFile) From 25c454b1a7da4855ead791484493021b9565602d Mon Sep 17 00:00:00 2001 From: kanthi subramanian Date: Tue, 18 Aug 2026 17:07:41 -0500 Subject: [PATCH 5/5] Fix CVE by pining httpclient to 5.4 --- ice-rest-catalog/pom.xml | 2 +- ice/pom.xml | 2 +- pom.xml | 17 +++++++++++++++++ 3 files changed, 19 insertions(+), 2 deletions(-) diff --git a/ice-rest-catalog/pom.xml b/ice-rest-catalog/pom.xml index aa42b90f..f735f49e 100644 --- a/ice-rest-catalog/pom.xml +++ b/ice-rest-catalog/pom.xml @@ -69,7 +69,7 @@ org.apache.httpcomponents.client5 httpclient5 - 5.4.3 + ${httpclient5.version} org.slf4j diff --git a/ice/pom.xml b/ice/pom.xml index 9a320f87..fe56ef4c 100644 --- a/ice/pom.xml +++ b/ice/pom.xml @@ -386,7 +386,7 @@ org.apache.httpcomponents.client5 httpclient5 - 5.4.3 + ${httpclient5.version} org.slf4j diff --git a/pom.xml b/pom.xml index ff8464e4..2455bcf3 100644 --- a/pom.xml +++ b/pom.xml @@ -34,6 +34,8 @@ 3.2.0 3.26.1 1.18.0 + 5.4.3 + 5.4.3 @@ -45,6 +47,21 @@ pom import + + org.apache.httpcomponents.client5 + httpclient5 + ${httpclient5.version} + + + org.apache.httpcomponents.core5 + httpcore5 + ${httpcore5.version} + + + org.apache.httpcomponents.core5 + httpcore5-h2 + ${httpcore5.version} +