From 1cd1803e0645a240bdac9cfcd32637deea11c42b Mon Sep 17 00:00:00 2001 From: Karthik Karuppiah Date: Mon, 24 Aug 2026 15:17:34 -0700 Subject: [PATCH 1/2] Add sources and javadoc jars to chalk-java-shaded publication Central requires sources and javadoc on every component, and both coordinates upload as a single deployment bundle -- so the incomplete chalk-java-shaded component failed validation and took chalk-java 1.3.3 down with it. Both publish runs (8/10, 8/14) uploaded successfully and were rejected server-side afterward, which is why CI stayed green with nothing on Central. Reuse the unshaded jars: relocation is a bytecode rewrite, so the sources are identical and the public ai.chalk.* API is unrelocated. Deferred to afterEvaluate because the vanniktech plugin registers both tasks after this file evaluates. --- build.gradle | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/build.gradle b/build.gradle index 9f91958..6ee6429 100644 --- a/build.gradle +++ b/build.gradle @@ -186,6 +186,17 @@ publishing { } } +// Central rejects any component missing sources/javadoc, and one bad component fails the whole +// deployment bundle — which would take chalk-java down with it. Reuse the unshaded jars: relocation +// is a bytecode rewrite, so the sources are identical and the public ai.chalk.* API is unrelocated. +// Deferred to afterEvaluate because the vanniktech plugin registers both tasks after this file runs. +afterEvaluate { + publishing.publications.named('shaded') { + artifact(tasks.named('sourcesJar')) { classifier = 'sources' } + artifact(tasks.named('mavenPlainJavadocJar')) { classifier = 'javadoc' } + } +} + // With two publications (maven + shaded), both sign tasks write their .asc into build/libs, and each // publish task reads from build/libs — so Gradle's strict validation flags an implicit dependency // between publish and the *other* publication's sign task and fails the build. Order all publish From af4637c028078b956af649cfee814ae10b3c4538 Mon Sep 17 00:00:00 2001 From: Karthik Karuppiah Date: Wed, 26 Aug 2026 11:55:31 -0700 Subject: [PATCH 2/2] Bound Flink sink retries by time, not attempt count Production Flink jobs restarted on transient UNAVAILABLE during query-server rollouts: "upload_features failed after 4 attempt(s)". The retries themselves worked, but a fixed maxRetries=3 with a 500ms base backoff only covers 0.5+1+2 = 3.5s -- shorter than a routine deploy, so the budget was spent while the backend was still coming back and the task failed. An attempt count can't express "survive a rollout": the real budget shifts whenever the backoff is tuned. Replace it with an explicit wall-clock budget, retryTimeout, defaulting to 120s. maxRetries stays as an opt-in hard cap and is now unlimited by default, so the time budget is the effective bound. Timing uses nanoTime so an NTP step can't move the deadline, and the loop stops when the next sleep would overrun it rather than sleeping past it. Retries still run inline on the task thread, so the budget is also the worst-case subtask stall; documented alongside the checkpoint-timeout interaction. --- docs/flink-sink.md | 8 ++- .../flink/core/ChalkFeatureUploader.java | 61 +++++++++++++------ .../ai/chalk/flink/core/ChalkSinkConfig.java | 31 +++++++++- .../flink/core/ChalkFeatureUploaderTest.java | 48 +++++++++++++++ .../chalk/flink/core/ChalkSinkConfigTest.java | 10 +++ 5 files changed, 136 insertions(+), 22 deletions(-) diff --git a/docs/flink-sink.md b/docs/flink-sink.md index f016a26..65219ef 100644 --- a/docs/flink-sink.md +++ b/docs/flink-sink.md @@ -105,12 +105,18 @@ Each element maps to `Map`. Rules: | `batchSize` | `1000` | Flush when this many rows are buffered. | | `flushInterval` | `5s` | Flush a non-empty buffer at least this often. | | `uploadTimeout` | `30s` | Per-call deadline. | -| `maxRetries` | `3` | Retries per batch on transient failure (exponential backoff). | +| `retryTimeout` | `120s` | Total wall-clock budget for retrying one batch's transient failures. The primary retry bound — sized to outlast a routine query-server rollout. | +| `maxRetries` | unlimited | Optional hard cap on retries per batch. Off by default so `retryTimeout` governs; set it to stop after a fixed number of attempts regardless of remaining budget. | | `retryBackoff` | `500ms` | Base backoff, doubled per attempt (capped 30s). | | `failOnUploadErrors` | `true` | `true`: fail the sink on engine data-level errors (replay). `false`: log a WARN and **drop** the rejected rows, continuing. | > **Poison-pill note:** engine *data-level* errors (e.g. a bad feature type) are deterministic — with `failOnUploadErrors=true`, replaying the same batch after a checkpoint restart hits the same error, wedging the pipeline in a restart loop. Fix such rows upstream, or set `failOnUploadErrors=false` to drop them. (Transient transport errors are handled separately by retry + Flink restart and do make progress.) +> **Retry budget:** retries block the task thread, so `retryTimeout` is also the worst-case stall for +> that subtask — keep it below the checkpoint timeout. An outage longer than the budget still surfaces +> as a task failure and a restart from the last checkpoint; that is at-least-once working as intended, +> not data loss. + ## Write targets `ChalkSinkConfig` exposes `writeOnline` (default `true`), `writeOffline`, and `updateMataggs` diff --git a/src/main/java/ai/chalk/flink/core/ChalkFeatureUploader.java b/src/main/java/ai/chalk/flink/core/ChalkFeatureUploader.java index ead8065..160d8c4 100644 --- a/src/main/java/ai/chalk/flink/core/ChalkFeatureUploader.java +++ b/src/main/java/ai/chalk/flink/core/ChalkFeatureUploader.java @@ -10,14 +10,20 @@ import java.util.LinkedHashSet; import java.util.List; import java.util.Map; +import java.util.concurrent.TimeUnit; /** * Flink-free batching engine for Chalk {@code upload_features}. Accumulates feature rows and flushes * them as a single columnar upload when the buffer fills (see {@link #add}), or when the owning sink * asks (checkpoint / close). Retries transient failures (gRPC UNAVAILABLE / DEADLINE_EXCEEDED - * / RESOURCE_EXHAUSTED / ABORTED) with exponential backoff; non-transient failures (auth, invalid - * argument) are thrown immediately. On exhaustion it throws, letting the caller (a Flink sink) fail - * and replay from the last checkpoint — at-least-once, made safe by upsert semantics. + * / RESOURCE_EXHAUSTED / ABORTED) with exponential backoff until the {@code retryTimeout} budget is + * spent; non-transient failures (auth, invalid argument) are thrown immediately. On exhaustion it + * throws, letting the caller (a Flink sink) fail and replay from the last checkpoint — at-least-once, + * made safe by upsert semantics. + * + *

The budget is time, not attempts, so it survives a backend rollout of a known duration + * regardless of how the backoff is tuned. Retries run inline on the task thread, so the budget is + * also the worst-case subtask stall. * *

Flush latency is bounded by {@code min(batchSize reached, checkpoint interval)}. The * configured flush interval is a best-effort upper bound evaluated when the next element arrives; on @@ -98,26 +104,39 @@ private boolean intervalElapsed() { } private UploadOutcome uploadWithRetry(Map> columnar) { - int attempts = config.maxRetries() + 1; + // nanoTime, not currentTimeMillis: the budget must not shift under an NTP step. + long start = System.nanoTime(); + long budgetNanos = TimeUnit.MILLISECONDS.toNanos(config.retryTimeoutMillis()); Exception last = null; - for (int attempt = 0; attempt < attempts; attempt++) { + int attempts = 0; + while (true) { try { return client.upload(columnar); } catch (Exception e) { last = e; + attempts++; // Fail fast on non-transient errors (auth, invalid argument, ...): retrying them - // just burns backoff and then triggers a Flink restart storm on the same error. - if (attempt >= attempts - 1 || !isRetryable(e)) { + // just burns the budget and then triggers a Flink restart storm on the same error. + if (!isRetryable(e) || attempts > config.maxRetries()) { + break; + } + long backoff = backoffMillis(attempts - 1); + // Stop once the next sleep would outlast the budget: sleeping past the deadline + // delays the restart without buying another attempt. + long remaining = TimeUnit.NANOSECONDS.toMillis(budgetNanos - (System.nanoTime() - start)); + if (backoff >= remaining) { break; } - LOG.warn("transient upload_features failure (attempt {}/{}), retrying: {}", - attempt + 1, attempts, e.toString()); - sleepBackoff(attempt); + LOG.warn("transient upload_features failure (attempt {}, {}ms of retry budget left), " + + "retrying in {}ms: {}", attempts, remaining, backoff, e.toString()); + sleep(backoff); } } - boolean retryable = isRetryable(last); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); throw new ChalkUploadException( - "upload_features failed" + (retryable ? " after " + attempts + " attempt(s)" + "upload_features failed" + (isRetryable(last) + ? " after " + attempts + " attempt(s) over " + elapsedMillis + "ms" + + " (retryTimeout=" + config.retryTimeoutMillis() + "ms)" : " with a non-retryable error"), last); } @@ -135,13 +154,19 @@ private static boolean isRetryable(Throwable t) { return false; } - private void sleepBackoff(int attempt) { - // Exponential backoff, capped at 30s to bound restart latency. Cap the shift and detect - // overflow (a huge maxRetries could otherwise wrap the product negative -> bad Thread.sleep). - long scaled = config.retryBackoffMillis() << Math.min(attempt, 20); - long backoff = (scaled < 0) ? 30_000L : Math.min(scaled, 30_000L); + /** + * Exponential backoff for the given zero-based retry index, capped at 30s so a long budget is + * spent on many attempts rather than a few enormous sleeps. The shift is capped and overflow + * detected: an unbounded attempt count could otherwise wrap the product negative. + */ + private long backoffMillis(int retryIndex) { + long scaled = config.retryBackoffMillis() << Math.min(retryIndex, 20); + return (scaled < 0) ? 30_000L : Math.min(scaled, 30_000L); + } + + private void sleep(long millis) { try { - Thread.sleep(backoff); + Thread.sleep(millis); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); throw new ChalkUploadException("interrupted while backing off before upload retry", ie); diff --git a/src/main/java/ai/chalk/flink/core/ChalkSinkConfig.java b/src/main/java/ai/chalk/flink/core/ChalkSinkConfig.java index 5bb3402..dc5f4b9 100644 --- a/src/main/java/ai/chalk/flink/core/ChalkSinkConfig.java +++ b/src/main/java/ai/chalk/flink/core/ChalkSinkConfig.java @@ -28,6 +28,7 @@ public final class ChalkSinkConfig implements Serializable { private final long uploadTimeoutMillis; private final int maxRetries; private final long retryBackoffMillis; + private final long retryTimeoutMillis; private final boolean failOnUploadErrors; private final boolean writeOnline; @@ -46,6 +47,7 @@ private ChalkSinkConfig(Builder b) { this.uploadTimeoutMillis = b.uploadTimeoutMillis; this.maxRetries = b.maxRetries; this.retryBackoffMillis = b.retryBackoffMillis; + this.retryTimeoutMillis = b.retryTimeoutMillis; this.failOnUploadErrors = b.failOnUploadErrors; this.writeOnline = b.writeOnline; this.writeOffline = b.writeOffline; @@ -63,6 +65,8 @@ private ChalkSinkConfig(Builder b) { public Duration uploadTimeout() { return Duration.ofMillis(uploadTimeoutMillis); } public int maxRetries() { return maxRetries; } public long retryBackoffMillis() { return retryBackoffMillis; } + public long retryTimeoutMillis() { return retryTimeoutMillis; } + public Duration retryTimeout() { return Duration.ofMillis(retryTimeoutMillis); } public boolean failOnUploadErrors() { return failOnUploadErrors; } public boolean writeOnline() { return writeOnline; } public boolean writeOffline() { return writeOffline; } @@ -82,8 +86,12 @@ public static final class Builder { private int batchSize = 1_000; private long flushIntervalMillis = 5_000; private long uploadTimeoutMillis = 30_000; - private int maxRetries = 3; + // Unlimited by default: retryTimeout is the real bound. A fixed attempt count can't express + // "survive a backend rollout" — with the default backoff, 3 retries covered only ~3.5s, which + // is shorter than a routine query-server deploy and caused Flink restarts in production. + private int maxRetries = Integer.MAX_VALUE; private long retryBackoffMillis = 500; + private long retryTimeoutMillis = 120_000; private boolean failOnUploadErrors = true; private boolean writeOnline = true; private boolean writeOffline = false; @@ -116,12 +124,26 @@ public static final class Builder { /** Per-call upload deadline. Default 30s. */ public Builder uploadTimeout(Duration v) { this.uploadTimeoutMillis = v.toMillis(); return this; } - /** Retries per batch on transient failures (in addition to the first attempt). Default 3. */ + /** + * Optional hard cap on retries per batch, in addition to the first attempt. Unlimited by + * default — {@link #retryTimeout} is the primary bound. Set this only to stop retrying after + * a fixed number of attempts regardless of how much time budget remains. + */ public Builder maxRetries(int v) { this.maxRetries = v; return this; } - /** Base backoff between retries, in millis (doubled each attempt). Default 500. */ + /** Base backoff between retries, in millis (doubled each attempt, capped at 30s). Default 500. */ public Builder retryBackoff(Duration v) { this.retryBackoffMillis = v.toMillis(); return this; } + /** + * Total wall-clock budget for retrying one batch's transient failures. Default 120s, chosen to + * outlast a routine query-server rollout: a shorter budget surfaces the outage as a Flink task + * failure and a restart from the last checkpoint. + * + *

Retries block the task thread, so this is also the worst-case stall for the subtask. + * Keep it below the checkpoint timeout. + */ + public Builder retryTimeout(Duration v) { this.retryTimeoutMillis = v.toMillis(); return this; } + /** * When true (default), a batch whose response carries engine data-level errors fails the sink. * @@ -171,6 +193,9 @@ public ChalkSinkConfig build() { if (retryBackoffMillis < 0) { throw new IllegalArgumentException("retryBackoff must be >= 0"); } + if (retryTimeoutMillis < 0) { + throw new IllegalArgumentException("retryTimeout must be >= 0"); + } // Fail fast at job-construction time (not later on the TaskManager) for write targets the // pinned chalk-java can't honor. See README "Write targets". if (writeOffline || updateMataggs || !writeOnline) { diff --git a/src/test/java/ai/chalk/flink/core/ChalkFeatureUploaderTest.java b/src/test/java/ai/chalk/flink/core/ChalkFeatureUploaderTest.java index 51bc910..9b90e14 100644 --- a/src/test/java/ai/chalk/flink/core/ChalkFeatureUploaderTest.java +++ b/src/test/java/ai/chalk/flink/core/ChalkFeatureUploaderTest.java @@ -115,6 +115,54 @@ void throwsAfterRetriesExhausted() { assertEquals(3, fake.attempts.get()); // 1 initial + 2 retries } + @Test + void stopsRetryingWhenTimeBudgetSpent() { + FakeClient fake = new FakeClient(); + fake.failFirstN = Integer.MAX_VALUE; // never recovers + ChalkFeatureUploader uploader = new ChalkFeatureUploader( + config(ChalkSinkConfig.builder().batchSize(1) + .retryTimeout(Duration.ofMillis(300)) + .retryBackoff(Duration.ofMillis(50))), fake); + + // maxRetries is unlimited by default, so the time budget is the only thing that can stop this. + long start = System.nanoTime(); + ChalkUploadException e = assertThrows(ChalkUploadException.class, + () -> uploader.add(Map.of("user.id", 1L))); + long elapsedMillis = (System.nanoTime() - start) / 1_000_000; + + assertTrue(e.getMessage().contains("retryTimeout"), e.getMessage()); + // Never sleeps past the deadline; generous ceiling so a slow CI box can't flake this. + assertTrue(elapsedMillis < 3_000, "took " + elapsedMillis + "ms, expected to stop near 300ms"); + } + + @Test + void timeBudgetAllowsMoreAttemptsThanTheOldFixedCap() { + FakeClient fake = new FakeClient(); + fake.failFirstN = Integer.MAX_VALUE; + ChalkFeatureUploader uploader = new ChalkFeatureUploader( + config(ChalkSinkConfig.builder().batchSize(1) + .retryTimeout(Duration.ofMillis(1_000)) + .retryBackoff(Duration.ofMillis(1))), fake); + + assertThrows(ChalkUploadException.class, () -> uploader.add(Map.of("user.id", 1L))); + // The old behaviour was a hard 4 attempts (maxRetries=3 + 1) regardless of remaining budget; + // with a 1s budget and 1ms base backoff there is room for many more. + assertTrue(fake.attempts.get() > 4, "only " + fake.attempts.get() + " attempts"); + } + + @Test + void explicitMaxRetriesStillCapsBeforeBudget() { + FakeClient fake = new FakeClient(); + fake.failFirstN = Integer.MAX_VALUE; + ChalkFeatureUploader uploader = new ChalkFeatureUploader( + config(ChalkSinkConfig.builder().batchSize(1).maxRetries(2) + .retryTimeout(Duration.ofMinutes(5)) // budget far larger than the cap + .retryBackoff(Duration.ofMillis(1))), fake); + + assertThrows(ChalkUploadException.class, () -> uploader.add(Map.of("user.id", 1L))); + assertEquals(3, fake.attempts.get()); // 1 initial + 2 retries, budget untouched + } + @Test void failsFastOnNonRetryableError() { FakeClient fake = new FakeClient(); diff --git a/src/test/java/ai/chalk/flink/core/ChalkSinkConfigTest.java b/src/test/java/ai/chalk/flink/core/ChalkSinkConfigTest.java index 175f647..4bd8468 100644 --- a/src/test/java/ai/chalk/flink/core/ChalkSinkConfigTest.java +++ b/src/test/java/ai/chalk/flink/core/ChalkSinkConfigTest.java @@ -5,6 +5,7 @@ import java.time.Duration; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; class ChalkSinkConfigTest { @@ -28,11 +29,20 @@ void rejectsUnsupportedWriteTargetsAtConstruction() { @Test void rejectsInvalidDurationsAndCounts() { assertThrows(IllegalArgumentException.class, () -> base().retryBackoff(Duration.ofMillis(-1)).build()); + assertThrows(IllegalArgumentException.class, () -> base().retryTimeout(Duration.ofMillis(-1)).build()); assertThrows(IllegalArgumentException.class, () -> base().flushInterval(Duration.ofMillis(-1)).build()); assertThrows(IllegalArgumentException.class, () -> base().uploadTimeout(Duration.ZERO).build()); assertThrows(IllegalArgumentException.class, () -> base().batchSize(0).build()); } + @Test + void retryBudgetDefaultsToTimeNotAttempts() { + ChalkSinkConfig c = base().build(); + assertEquals(Duration.ofSeconds(120), c.retryTimeout()); + // Unlimited by default so retryTimeout is the effective bound. + assertEquals(Integer.MAX_VALUE, c.maxRetries()); + } + @Test void rejectsMissingCredentials() { assertThrows(IllegalArgumentException.class,