diff --git a/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/ConsumerConfig.java b/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/ConsumerConfig.java new file mode 100644 index 00000000..0a215cf6 --- /dev/null +++ b/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/ConsumerConfig.java @@ -0,0 +1,198 @@ +package io.github.eschizoid.kpipe; + +import io.github.eschizoid.kpipe.consumer.CircuitBreakerController; +import io.github.eschizoid.kpipe.consumer.KPipeConsumer; +import io.github.eschizoid.kpipe.consumer.KPipeConsumerBuilder; +import io.github.eschizoid.kpipe.consumer.ProcessingMode; +import io.github.eschizoid.kpipe.metrics.ConsumerMetrics; +import io.github.eschizoid.kpipe.producer.tracing.Tracer; +import java.time.Duration; +import java.util.List; +import java.util.function.BiFunction; +import java.util.function.Consumer; +import java.util.function.Predicate; + +/// Package-private immutable holder for every consumer-wide setting the fluent facade exposes: +/// retry, backpressure, processing mode, key-ordered cap, metrics, error handler, dead-letter +/// topic, poll timeout, tracer, and circuit breaker. One `KPipeConsumer` carries one of these, +/// whether it was built from a single [Stream] or folded out of N [MultiBuilder] routes. +/// +/// This type is the single registration point for a consumer-wide setting. Before it existed the +/// same list was mirrored across five sites (the [DefaultStream] components + withers, its apply +/// chain, the [MultiBuilder] fields + withers, its re-implemented apply chain, and the per-route +/// rejection checks), and the mirrors drifted — the batch path once silently dropped tracer and +/// circuit breaker. Adding a setting now means: a component here (+ [Mut] line), one line in +/// [#applyTo], one [#CONSUMER_WIDE_SETTINGS] descriptor, and the thin `with*` delegates on +/// [Stream]/[MultiBuilder]. +/// +/// Nullability encodes "unset": reference components left `null` (and `maxRetries == 0`, +/// `processingMode == PARALLEL`, `keyOrderedMaxKeys == DEFAULT_KEY_ORDERED_MAX_KEYS`) mean the +/// underlying builder keeps its own default. +record ConsumerConfig( + int maxRetries, + Duration retryBackoff, + Long backpressureHigh, + Long backpressureLow, + ProcessingMode processingMode, + int keyOrderedMaxKeys, + ConsumerMetrics consumerMetrics, + Consumer errorHandler, + String deadLetterTopic, + Duration pollTimeout, + Tracer tracer, + CircuitBreakerController circuitBreaker +) { + /// The all-unset configuration: no retry, no backpressure, parallel mode with the default + /// key-ordered cap, and every optional component `null`. + static ConsumerConfig defaults() { + return new ConsumerConfig( + 0, + Duration.ofMillis(500), + null, + null, + ProcessingMode.PARALLEL, + ProcessingMode.DEFAULT_KEY_ORDERED_MAX_KEYS, + null, + null, + null, + null, + null, + null + ); + } + + /// Single funnel for updates: snapshot into a [Mut], let the caller change what they need, + /// rebuild a new immutable record. Same shape as `DefaultStream.mutate`. + ConsumerConfig with(final Consumer change) { + final var m = Mut.from(this); + change.accept(m); + return m.build(); + } + + /// Applies every set (non-default) setting onto `builder`. This is THE apply chain — both + /// single-stream sinks ([DefaultSink] / [DefaultBatchSink]) and [MultiBuilder#start()] call it, + /// so a new setting wired here reaches all three paths at once. Setter order is irrelevant to + /// the builder (all cross-setting derivation happens in its `build()`), so one fixed order + /// serves every caller. + void applyTo(final KPipeConsumerBuilder builder) { + builder.withProcessingMode(processingMode); + builder.withKeyOrderedMaxKeys(keyOrderedMaxKeys); + if (maxRetries > 0) builder.withRetry(maxRetries, retryBackoff); + if (backpressureHigh != null) builder.withBackpressure(backpressureHigh, backpressureLow); + if (consumerMetrics != null) builder.withMetrics(consumerMetrics); + if (errorHandler != null) builder.withErrorHandler(errorHandler::accept); + if (deadLetterTopic != null) builder.withDeadLetterTopic(deadLetterTopic); + if (pollTimeout != null) builder.withPollTimeout(pollTimeout); + if (tracer != null) builder.withTracer(tracer); + if (circuitBreaker != null) builder.withCircuitBreaker(circuitBreaker); + } + + /// One consumer-wide setting as seen by the [MultiBuilder] per-route guard: the `Stream.with*` + /// name, a predicate telling whether a route's config sets it, and the rejection message + /// pointing the user at the symmetric `MultiBuilder.with*` mirror. + record ConsumerWideSetting( + String setting, + Predicate isSet, + BiFunction rejection + ) {} + + /// Descriptor per consumer-wide setting, iterated by + /// `MultiBuilder.rejectPerRouteConsumerWideSettings` instead of one hand-written check per + /// setting. A setting registered here can never be silently dropped by a route configurator. + /// Order is the reporting order when a route sets several at once. + static final List CONSUMER_WIDE_SETTINGS = List.of( + new ConsumerWideSetting( + "withProcessingMode", + c -> c.processingMode() != ProcessingMode.PARALLEL, + (topic, c) -> + "Route '%s' sets withProcessingMode(%s) on its Stream, but processing mode is a consumer-wide setting. ".formatted( + topic, + c.processingMode() + ) + + "Move the call to MultiBuilder.withProcessingMode(...) instead." + ), + new ConsumerWideSetting( + "withKeyOrderedMaxKeys", + c -> c.keyOrderedMaxKeys() != ProcessingMode.DEFAULT_KEY_ORDERED_MAX_KEYS, + (topic, c) -> + "Route '%s' sets withKeyOrderedMaxKeys(%d) on its Stream, but the key-ordered key cap is a consumer-wide setting. ".formatted( + topic, + c.keyOrderedMaxKeys() + ) + + "Move the call to MultiBuilder.withKeyOrderedMaxKeys(...) instead." + ), + mirrored("withMetrics", c -> c.consumerMetrics() != null), + mirrored("withTracer", c -> c.tracer() != null), + mirrored("withCircuitBreaker", c -> c.circuitBreaker() != null), + mirrored("withRetry", c -> c.maxRetries() > 0), + mirrored("withBackpressure", c -> c.backpressureHigh() != null), + mirrored("withDeadLetterTopic", c -> c.deadLetterTopic() != null), + mirrored("withErrorHandler", c -> c.errorHandler() != null), + mirrored("withPollTimeout", c -> c.pollTimeout() != null) + ); + + /// Descriptor for the common case: a setting whose rejection message points at the + /// `MultiBuilder` mirror method of the same name. + private static ConsumerWideSetting mirrored(final String setting, final Predicate isSet) { + final var mirror = "MultiBuilder.%s(...)".formatted(setting); + return new ConsumerWideSetting( + setting, + isSet, + (topic, c) -> + "Route '%s' sets %s on its Stream, but %s is a consumer-wide setting; ".formatted(topic, setting, setting) + + "set it on %s instead.".formatted(mirror) + ); + } + + /// Mutable mirror of [ConsumerConfig]'s components used only inside [#with]. Never escapes the + /// package. + static final class Mut { + + int maxRetries; + Duration retryBackoff; + Long backpressureHigh; + Long backpressureLow; + ProcessingMode processingMode; + int keyOrderedMaxKeys; + ConsumerMetrics consumerMetrics; + Consumer errorHandler; + String deadLetterTopic; + Duration pollTimeout; + Tracer tracer; + CircuitBreakerController circuitBreaker; + + static Mut from(final ConsumerConfig c) { + final var m = new Mut(); + m.maxRetries = c.maxRetries; + m.retryBackoff = c.retryBackoff; + m.backpressureHigh = c.backpressureHigh; + m.backpressureLow = c.backpressureLow; + m.processingMode = c.processingMode; + m.keyOrderedMaxKeys = c.keyOrderedMaxKeys; + m.consumerMetrics = c.consumerMetrics; + m.errorHandler = c.errorHandler; + m.deadLetterTopic = c.deadLetterTopic; + m.pollTimeout = c.pollTimeout; + m.tracer = c.tracer; + m.circuitBreaker = c.circuitBreaker; + return m; + } + + ConsumerConfig build() { + return new ConsumerConfig( + maxRetries, + retryBackoff, + backpressureHigh, + backpressureLow, + processingMode, + keyOrderedMaxKeys, + consumerMetrics, + errorHandler, + deadLetterTopic, + pollTimeout, + tracer, + circuitBreaker + ); + } + } +} diff --git a/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultBatchSink.java b/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultBatchSink.java index eebc2c94..30ed6524 100644 --- a/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultBatchSink.java +++ b/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultBatchSink.java @@ -12,10 +12,9 @@ /// [io.github.eschizoid.kpipe.registry.MessagePipeline] /// from the stream's operator chain (deserialize → operators → return value) and wires it into /// [KPipeConsumerBuilder#withBatchPipeline] together with the configured [BatchSink] and -/// [BatchPolicy]. Honors whatever processing mode the underlying [DefaultStream] carries -/// (default: parallel); consumer-level config (metrics, errorHandler, DLQ, pollTimeout, retry, -/// backpressure, tracer, circuit breaker) carries over too via -/// [DefaultStream#applyCommonConsumerConfig]. +/// [BatchPolicy]. Every consumer-wide setting the underlying [DefaultStream] carries (processing +/// mode, metrics, errorHandler, DLQ, pollTimeout, retry, backpressure, tracer, circuit breaker) +/// carries over via [ConsumerConfig#applyTo] — the same chain the non-batch path uses. /// /// `Stream.toBatch(...)` produces a single-topic instance. [MultiBuilder] also constructs single- /// topic instances per route and collects them at `start()` time into one consumer subscribing @@ -84,11 +83,9 @@ public Handle start() { ); final var consumerBuilder = KPipeConsumer.builder() .withProperties(stream.kafkaProps()) - .withProcessingMode(stream.processingMode()) - .withKeyOrderedMaxKeys(stream.keyOrderedMaxKeys()) .withBatchPipeline(topic(), buildPipeline(), batchSink, batchPolicy); - stream.applyCommonConsumerConfig(consumerBuilder); + stream.consumerConfig().applyTo(consumerBuilder); return DefaultHandle.startAndWrap(consumerBuilder.build()); } diff --git a/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultSink.java b/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultSink.java index b9b25c74..1ca96456 100644 --- a/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultSink.java +++ b/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultSink.java @@ -36,11 +36,9 @@ public Handle start() { final var consumerBuilder = KPipeConsumer.builder() .withProperties(stream.kafkaProps()) .withTopics(stream.topics()) - .withPipeline(buildPipeline()) - .withProcessingMode(stream.processingMode()) - .withKeyOrderedMaxKeys(stream.keyOrderedMaxKeys()); + .withPipeline(buildPipeline()); - stream.applyCommonConsumerConfig(consumerBuilder); + stream.consumerConfig().applyTo(consumerBuilder); return DefaultHandle.startAndWrap(consumerBuilder.build()); } diff --git a/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultStream.java b/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultStream.java index 11b0ae3d..500bd1a4 100644 --- a/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultStream.java +++ b/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultStream.java @@ -3,7 +3,6 @@ import io.github.eschizoid.kpipe.consumer.BackpressureController; import io.github.eschizoid.kpipe.consumer.CircuitBreakerController; import io.github.eschizoid.kpipe.consumer.KPipeConsumer; -import io.github.eschizoid.kpipe.consumer.KPipeConsumerBuilder; import io.github.eschizoid.kpipe.consumer.ProcessingMode; import io.github.eschizoid.kpipe.format.avro.AvroFormat; import io.github.eschizoid.kpipe.format.protobuf.ProtobufFormat; @@ -37,10 +36,11 @@ /// This makes branching safe (`s.pipe(a)` and `s.pipe(b)` produce independent streams) and /// matches the Java Stream API's functional style. /// -/// Adding a new fluent setter is a one-place change: declare the record component, mirror it on -/// [Mut], and write the `with*` method as `mutate(m -> m.field = newValue)`. There is no -/// constructor copy-paste — [#mutate(Consumer)] funnels every wither through a single instance -/// of [Mut] and rebuilds via [Mut#build]. +/// Pipeline-level state (topics, format, operators, skipBytes, result observers) lives as record +/// components here; every consumer-wide setting lives in the single [ConsumerConfig] component, +/// whose `with*` methods below are thin delegates. Adding a new fluent setter is a one-place +/// change: declare it on [ConsumerConfig] (or as a component + [Mut] field here if it is +/// pipeline-level) and write the `with*` method as a `mutate(...)` / `mutateConfig(...)` lambda. /// /// @param the deserialized message type record DefaultStream( @@ -49,19 +49,8 @@ record DefaultStream( MessageFormat format, Supplier> defaultConsoleSinkFactory, List> operators, - int maxRetries, - Duration retryBackoff, - Long backpressureHigh, - Long backpressureLow, - ProcessingMode processingMode, - int keyOrderedMaxKeys, int skipBytes, - ConsumerMetrics consumerMetrics, - Consumer errorHandler, - String deadLetterTopic, - Duration pollTimeout, - Tracer tracer, - CircuitBreakerController circuitBreaker, + ConsumerConfig consumerConfig, Runnable onFilteredObserver, Consumer onFailedObserver, Consumer> peekResultObserver @@ -95,18 +84,7 @@ record DefaultStream( Objects.requireNonNull(defaultConsoleSinkFactory, "defaultConsoleSinkFactory cannot be null"), List.of(), 0, - Duration.ofMillis(500), - null, - null, - ProcessingMode.PARALLEL, - ProcessingMode.DEFAULT_KEY_ORDERED_MAX_KEYS, - 0, - null, - null, - null, - null, - null, - null, + ConsumerConfig.defaults(), null, null, null @@ -151,9 +129,9 @@ public Stream when(final Predicate cond, final UnaryOperator ifTrue, fi public Stream withRetry(final int maxRetries, final Duration backoff) { if (maxRetries < 0) throw new IllegalArgumentException("maxRetries cannot be negative"); Objects.requireNonNull(backoff, "backoff cannot be null"); - return mutate(m -> { - m.maxRetries = maxRetries; - m.retryBackoff = backoff; + return mutateConfig(c -> { + c.maxRetries = maxRetries; + c.retryBackoff = backoff; }); } @@ -170,29 +148,31 @@ public Stream withBackpressure(final long high, final long low) { if (low < 0 || low >= high) throw new IllegalArgumentException( "withBackpressure requires high > low > 0 (got high=%d, low=%d)".formatted(high, low) ); - return mutate(m -> { - m.backpressureHigh = high; - m.backpressureLow = low; + return mutateConfig(c -> { + c.backpressureHigh = high; + c.backpressureLow = low; }); } @Override public Stream withProcessingMode(final ProcessingMode mode) { Objects.requireNonNull(mode, "mode cannot be null"); - return mutate(m -> m.processingMode = mode); + return mutateConfig(c -> c.processingMode = mode); } @Override public Stream withKeyOrderedMaxKeys(final int maxKeys) { if (maxKeys <= 0) throw new IllegalArgumentException("maxKeys must be positive, got " + maxKeys); - return mutate(m -> m.keyOrderedMaxKeys = maxKeys); + return mutateConfig(c -> c.keyOrderedMaxKeys = maxKeys); } @Override public Stream skipBytes(final int n) { if (n < 0) throw new IllegalArgumentException("n cannot be negative"); if (n > 0 && format.isSchemaRegistryBacked()) throw new IllegalArgumentException( - "skipBytes(" + n + ") cannot be combined with a Schema-Registry-backed format: the format " + + "skipBytes(" + + n + + ") cannot be combined with a Schema-Registry-backed format: the format " + "reads the Confluent wire envelope itself, and stripping bytes first would corrupt every " + "record. Drop the skipBytes(...) call." ); @@ -203,7 +183,9 @@ public Stream skipBytes(final int n) { public Stream withSchemaRegistry(final SchemaResolver resolver) { Objects.requireNonNull(resolver, "resolver cannot be null"); if (skipBytes > 0) throw new IllegalArgumentException( - "withSchemaRegistry(...) cannot be combined with skipBytes(" + skipBytes + "): the " + + "withSchemaRegistry(...) cannot be combined with skipBytes(" + + skipBytes + + "): the " + "registry-backed format reads the Confluent wire envelope itself, and stripping bytes " + "first would corrupt every record. Drop the skipBytes(...) call." ); @@ -228,31 +210,31 @@ private static MessageFormat registryBackedFormat(final MessageFormat form @Override public Stream withMetrics(final ConsumerMetrics metrics) { Objects.requireNonNull(metrics, "metrics cannot be null"); - return mutate(m -> m.consumerMetrics = metrics); + return mutateConfig(c -> c.consumerMetrics = metrics); } @Override public Stream withErrorHandler(final Consumer handler) { Objects.requireNonNull(handler, "handler cannot be null"); - return mutate(m -> m.errorHandler = handler); + return mutateConfig(c -> c.errorHandler = handler); } @Override public Stream withDeadLetterTopic(final String dlqTopic) { if (dlqTopic == null || dlqTopic.isBlank()) throw new IllegalArgumentException("dlqTopic cannot be null or blank"); - return mutate(m -> m.deadLetterTopic = dlqTopic); + return mutateConfig(c -> c.deadLetterTopic = dlqTopic); } @Override public Stream withPollTimeout(final Duration timeout) { Objects.requireNonNull(timeout, "timeout cannot be null"); - return mutate(m -> m.pollTimeout = timeout); + return mutateConfig(c -> c.pollTimeout = timeout); } @Override public Stream withTracer(final Tracer tracer) { Objects.requireNonNull(tracer, "tracer cannot be null"); - return mutate(m -> m.tracer = tracer); + return mutateConfig(c -> c.tracer = tracer); } @Override @@ -267,7 +249,7 @@ public Stream withCircuitBreaker( @Override public Stream withCircuitBreaker(final CircuitBreakerController controller) { Objects.requireNonNull(controller, "controller cannot be null"); - return mutate(m -> m.circuitBreaker = controller); + return mutateConfig(c -> c.circuitBreaker = controller); } @Override @@ -322,27 +304,10 @@ private DefaultStream withOperator(final UnaryOperator op) { }); } - /// Applies every consumer-level setting that both [DefaultSink] and [DefaultBatchSink] share onto - /// `builder`: retry, backpressure, metrics, error handler, dead-letter topic, poll timeout, - /// tracer, and circuit breaker. All are consumer-wide, so wiring them from one place keeps the - /// two sink types from drifting — the batch path previously skipped tracer and circuit breaker, - /// silently dropping both when set on a `toBatch(...)` stream. - void applyCommonConsumerConfig(final KPipeConsumerBuilder builder) { - if (maxRetries > 0) builder.withRetry(maxRetries, retryBackoff); - if (backpressureHigh != null) builder.withBackpressure(backpressureHigh, backpressureLow); - if (consumerMetrics != null) builder.withMetrics(consumerMetrics); - if (errorHandler != null) builder.withErrorHandler(errorHandler::accept); - if (deadLetterTopic != null) builder.withDeadLetterTopic(deadLetterTopic); - if (pollTimeout != null) builder.withPollTimeout(pollTimeout); - if (tracer != null) builder.withTracer(tracer); - if (circuitBreaker != null) builder.withCircuitBreaker(circuitBreaker); - } - /// Wraps `base` with dispatch to the configured result observers (`onFiltered` / `onFailed` / - /// `peekResult`), or returns `base` unchanged when none is set. Lives here — next to - /// [#applyCommonConsumerConfig], and for the same reason — so both [DefaultSink] and - /// [DefaultBatchSink] share one wiring site: the batch path previously had no observer dispatch - /// at all, silently dropping observers set on a `toBatch(...)` stream. + /// `peekResult`), or returns `base` unchanged when none is set. Lives here so both + /// [DefaultSink] and [DefaultBatchSink] share one wiring site: the batch path previously had + /// no observer dispatch at all, silently dropping observers set on a `toBatch(...)` stream. /// /// Observers fire on the PIPELINE outcome at `process()` time — for the batch path that is /// before the record is buffered; batch-sink failures are a separate concern routed through the @@ -404,14 +369,20 @@ private static void safeAccept(final Consumer observer, final A arg, fina } /// Single funnel for every wither: snapshot this record into a [Mut], let the caller change - /// what they need, rebuild a new record. Replaces 13 hand-rolled 15-arg constructor calls with - /// one. New fields slot in at one site (the Mut declaration + its `from`/`build`). + /// what they need, rebuild a new record. New pipeline-level fields slot in at one site (the + /// Mut declaration + its `from`/`build`); consumer-wide fields go through [#mutateConfig]. private DefaultStream mutate(final Consumer> change) { final var m = Mut.from(this); change.accept(m); return m.build(); } + /// Funnel for the consumer-wide withers: rebuilds this stream around an updated + /// [ConsumerConfig], preserving the immutability contract (a new stream per call). + private DefaultStream mutateConfig(final Consumer change) { + return mutate(m -> m.consumerConfig = consumerConfig.with(change)); + } + /// Mutable mirror of [DefaultStream]'s components used only inside [#mutate]. Each public /// wither hands a freshly-allocated `Mut` to a small lambda that updates one or two fields, /// then [#build] returns a new immutable record. Never escapes the package. @@ -422,19 +393,8 @@ private static final class Mut { MessageFormat format; Supplier> defaultConsoleSinkFactory; List> operators; - int maxRetries; - Duration retryBackoff; - Long backpressureHigh; - Long backpressureLow; - ProcessingMode processingMode; - int keyOrderedMaxKeys; int skipBytes; - ConsumerMetrics consumerMetrics; - Consumer errorHandler; - String deadLetterTopic; - Duration pollTimeout; - Tracer tracer; - CircuitBreakerController circuitBreaker; + ConsumerConfig consumerConfig; Runnable onFilteredObserver; Consumer onFailedObserver; Consumer> peekResultObserver; @@ -446,19 +406,8 @@ static Mut from(final DefaultStream s) { m.format = s.format; m.defaultConsoleSinkFactory = s.defaultConsoleSinkFactory; m.operators = s.operators; - m.maxRetries = s.maxRetries; - m.retryBackoff = s.retryBackoff; - m.backpressureHigh = s.backpressureHigh; - m.backpressureLow = s.backpressureLow; - m.processingMode = s.processingMode; - m.keyOrderedMaxKeys = s.keyOrderedMaxKeys; m.skipBytes = s.skipBytes; - m.consumerMetrics = s.consumerMetrics; - m.errorHandler = s.errorHandler; - m.deadLetterTopic = s.deadLetterTopic; - m.pollTimeout = s.pollTimeout; - m.tracer = s.tracer; - m.circuitBreaker = s.circuitBreaker; + m.consumerConfig = s.consumerConfig; m.onFilteredObserver = s.onFilteredObserver; m.onFailedObserver = s.onFailedObserver; m.peekResultObserver = s.peekResultObserver; @@ -472,19 +421,8 @@ DefaultStream build() { format, defaultConsoleSinkFactory, operators, - maxRetries, - retryBackoff, - backpressureHigh, - backpressureLow, - processingMode, - keyOrderedMaxKeys, skipBytes, - consumerMetrics, - errorHandler, - deadLetterTopic, - pollTimeout, - tracer, - circuitBreaker, + consumerConfig, onFilteredObserver, onFailedObserver, peekResultObserver diff --git a/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/MultiBuilder.java b/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/MultiBuilder.java index 703df3dd..8a754164 100644 --- a/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/MultiBuilder.java +++ b/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/MultiBuilder.java @@ -50,18 +50,7 @@ public final class MultiBuilder { private final Properties kafkaProps; private final Map> routes = new LinkedHashMap<>(); - private ConsumerMetrics consumerMetrics; - private Tracer tracer; - private CircuitBreakerController circuitBreaker; - private ProcessingMode processingMode = ProcessingMode.PARALLEL; - private Integer keyOrderedMaxKeys; - private Integer maxRetries; - private Duration retryBackoff; - private Long backpressureHigh; - private Long backpressureLow; - private Consumer errorHandler; - private String deadLetterTopic; - private Duration pollTimeout; + private ConsumerConfig consumerConfig = ConsumerConfig.defaults(); MultiBuilder(final Properties kafkaProps) { this.kafkaProps = (Properties) Objects.requireNonNull(kafkaProps, "kafkaProps cannot be null").clone(); @@ -75,7 +64,8 @@ public final class MultiBuilder { /// from `kpipe-metrics-otel`) /// @return this builder public MultiBuilder withMetrics(final ConsumerMetrics metrics) { - this.consumerMetrics = Objects.requireNonNull(metrics, "metrics cannot be null"); + Objects.requireNonNull(metrics, "metrics cannot be null"); + consumerConfig = consumerConfig.with(c -> c.consumerMetrics = metrics); return this; } @@ -86,7 +76,8 @@ public MultiBuilder withMetrics(final ConsumerMetrics metrics) { /// `kpipe-tracing-otel`); pass `Tracer.noop()` to disable explicitly /// @return this builder public MultiBuilder withTracer(final Tracer tracer) { - this.tracer = Objects.requireNonNull(tracer, "tracer cannot be null"); + Objects.requireNonNull(tracer, "tracer cannot be null"); + consumerConfig = consumerConfig.with(c -> c.tracer = tracer); return this; } @@ -97,7 +88,8 @@ public MultiBuilder withTracer(final Tracer tracer) { /// @param controller the breaker policy (must not be null) /// @return this builder public MultiBuilder withCircuitBreaker(final CircuitBreakerController controller) { - this.circuitBreaker = Objects.requireNonNull(controller, "controller cannot be null"); + Objects.requireNonNull(controller, "controller cannot be null"); + consumerConfig = consumerConfig.with(c -> c.circuitBreaker = controller); return this; } @@ -110,7 +102,8 @@ public MultiBuilder withCircuitBreaker(final CircuitBreakerController controller /// @param mode the processing mode (must not be null) /// @return this builder public MultiBuilder withProcessingMode(final ProcessingMode mode) { - this.processingMode = Objects.requireNonNull(mode, "mode cannot be null"); + Objects.requireNonNull(mode, "mode cannot be null"); + consumerConfig = consumerConfig.with(c -> c.processingMode = mode); return this; } @@ -121,7 +114,7 @@ public MultiBuilder withProcessingMode(final ProcessingMode mode) { /// @return this builder public MultiBuilder withKeyOrderedMaxKeys(final int maxKeys) { if (maxKeys <= 0) throw new IllegalArgumentException("maxKeys must be positive, got " + maxKeys); - this.keyOrderedMaxKeys = maxKeys; + consumerConfig = consumerConfig.with(c -> c.keyOrderedMaxKeys = maxKeys); return this; } @@ -134,8 +127,11 @@ public MultiBuilder withKeyOrderedMaxKeys(final int maxKeys) { /// @return this builder public MultiBuilder withRetry(final int maxRetries, final Duration backoff) { if (maxRetries < 0) throw new IllegalArgumentException("maxRetries cannot be negative, got " + maxRetries); - this.maxRetries = maxRetries; - this.retryBackoff = Objects.requireNonNull(backoff, "backoff cannot be null"); + Objects.requireNonNull(backoff, "backoff cannot be null"); + consumerConfig = consumerConfig.with(c -> { + c.maxRetries = maxRetries; + c.retryBackoff = backoff; + }); return this; } @@ -162,8 +158,10 @@ public MultiBuilder withBackpressure(final long high, final long low) { if (low < 0 || low >= high) throw new IllegalArgumentException( "withBackpressure requires high > low > 0 (got high=%d, low=%d)".formatted(high, low) ); - this.backpressureHigh = high; - this.backpressureLow = low; + consumerConfig = consumerConfig.with(c -> { + c.backpressureHigh = high; + c.backpressureLow = low; + }); return this; } @@ -173,7 +171,8 @@ public MultiBuilder withBackpressure(final long high, final long low) { /// @param handler the error callback (must be non-null) /// @return this builder public MultiBuilder withErrorHandler(final Consumer handler) { - this.errorHandler = Objects.requireNonNull(handler, "handler cannot be null"); + Objects.requireNonNull(handler, "handler cannot be null"); + consumerConfig = consumerConfig.with(c -> c.errorHandler = handler); return this; } @@ -186,7 +185,7 @@ public MultiBuilder withDeadLetterTopic(final String dlqTopic) { if (dlqTopic == null || dlqTopic.isBlank()) throw new IllegalArgumentException( "dlqTopic cannot be null or blank, got '" + dlqTopic + "'" ); - this.deadLetterTopic = dlqTopic; + consumerConfig = consumerConfig.with(c -> c.deadLetterTopic = dlqTopic); return this; } @@ -196,7 +195,8 @@ public MultiBuilder withDeadLetterTopic(final String dlqTopic) { /// @param timeout the poll timeout (must be non-null) /// @return this builder public MultiBuilder withPollTimeout(final Duration timeout) { - this.pollTimeout = Objects.requireNonNull(timeout, "timeout cannot be null"); + Objects.requireNonNull(timeout, "timeout cannot be null"); + consumerConfig = consumerConfig.with(c -> c.pollTimeout = timeout); return this; } @@ -363,16 +363,7 @@ public Handle start() { } if (!nonBatchPipelines.isEmpty()) consumerBuilder.withPipelines(nonBatchPipelines); - consumerBuilder.withProcessingMode(processingMode); - if (keyOrderedMaxKeys != null) consumerBuilder.withKeyOrderedMaxKeys(keyOrderedMaxKeys); - if (consumerMetrics != null) consumerBuilder.withMetrics(consumerMetrics); - if (tracer != null) consumerBuilder.withTracer(tracer); - if (circuitBreaker != null) consumerBuilder.withCircuitBreaker(circuitBreaker); - if (maxRetries != null && maxRetries > 0) consumerBuilder.withRetry(maxRetries, retryBackoff); - if (backpressureHigh != null) consumerBuilder.withBackpressure(backpressureHigh, backpressureLow); - if (errorHandler != null) consumerBuilder.withErrorHandler(errorHandler::accept); - if (deadLetterTopic != null) consumerBuilder.withDeadLetterTopic(deadLetterTopic); - if (pollTimeout != null) consumerBuilder.withPollTimeout(pollTimeout); + consumerConfig.applyTo(consumerBuilder); return DefaultHandle.startAndWrap(consumerBuilder.build()); } @@ -384,58 +375,20 @@ public Handle start() { /// /// Routes are expected to be terminal sinks built via `toCustom(...)` / `toBatch(...)` / /// `toConsole()` etc.; unknown sink shapes are passed through (no false-positives). + /// + /// Iterates [ConsumerConfig#CONSUMER_WIDE_SETTINGS] rather than one hand-written check per + /// setting — registering a setting on the descriptor list is what makes this guard cover it. private static void rejectPerRouteConsumerWideSettings(final String topic, final Sink sink) { final DefaultStream stream; if (sink instanceof DefaultSink ds) stream = ds.stream(); else if (sink instanceof DefaultBatchSink dbs) stream = dbs.stream(); else return; - if (stream.processingMode() != ProcessingMode.PARALLEL) throw new IllegalArgumentException( - "Route '%s' sets withProcessingMode(%s) on its Stream, but processing mode is a consumer-wide setting. ".formatted( - topic, - stream.processingMode() - ) + - "Move the call to MultiBuilder.withProcessingMode(...) instead." - ); - if (stream.keyOrderedMaxKeys() != ProcessingMode.DEFAULT_KEY_ORDERED_MAX_KEYS) throw new IllegalArgumentException( - "Route '%s' sets withKeyOrderedMaxKeys(%d) on its Stream, but the key-ordered key cap is a consumer-wide setting. ".formatted( - topic, - stream.keyOrderedMaxKeys() - ) + - "Move the call to MultiBuilder.withKeyOrderedMaxKeys(...) instead." - ); - if (stream.consumerMetrics() != null) throw new IllegalArgumentException( - perRouteRejection(topic, "withMetrics", "MultiBuilder.withMetrics(...)") - ); - if (stream.tracer() != null) throw new IllegalArgumentException( - perRouteRejection(topic, "withTracer", "MultiBuilder.withTracer(...)") - ); - if (stream.circuitBreaker() != null) throw new IllegalArgumentException( - perRouteRejection(topic, "withCircuitBreaker", "MultiBuilder.withCircuitBreaker(...)") - ); - if (stream.maxRetries() > 0) throw new IllegalArgumentException( - perRouteRejection(topic, "withRetry", "MultiBuilder.withRetry(...)") - ); - if (stream.backpressureHigh() != null) throw new IllegalArgumentException( - perRouteRejection(topic, "withBackpressure", "MultiBuilder.withBackpressure(...)") - ); - if (stream.deadLetterTopic() != null) throw new IllegalArgumentException( - perRouteRejection(topic, "withDeadLetterTopic", "MultiBuilder.withDeadLetterTopic(...)") - ); - if (stream.errorHandler() != null) throw new IllegalArgumentException( - perRouteRejection(topic, "withErrorHandler", "MultiBuilder.withErrorHandler(...)") - ); - if (stream.pollTimeout() != null) throw new IllegalArgumentException( - perRouteRejection(topic, "withPollTimeout", "MultiBuilder.withPollTimeout(...)") - ); - } - - /// Builds the rejection message for a per-route setting that already has a symmetric - /// `MultiBuilder.with*` setter — point the user at it. - private static String perRouteRejection(final String topic, final String setting, final String mirror) { - return ( - "Route '%s' sets %s on its Stream, but %s is a consumer-wide setting; ".formatted(topic, setting, setting) + - "set it on %s instead.".formatted(mirror) - ); + final var routeConfig = stream.consumerConfig(); + for (final var setting : ConsumerConfig.CONSUMER_WIDE_SETTINGS) { + if (setting.isSet().test(routeConfig)) throw new IllegalArgumentException( + setting.rejection().apply(topic, routeConfig) + ); + } } /// Type witness: pulls the typed pipeline + sink off the route, then calls the typed builder diff --git a/lib/kpipe-api/src/test/java/io/github/eschizoid/kpipe/KPipeFacadeBuildTest.java b/lib/kpipe-api/src/test/java/io/github/eschizoid/kpipe/KPipeFacadeBuildTest.java index 58289ba8..b2cb80aa 100644 --- a/lib/kpipe-api/src/test/java/io/github/eschizoid/kpipe/KPipeFacadeBuildTest.java +++ b/lib/kpipe-api/src/test/java/io/github/eschizoid/kpipe/KPipeFacadeBuildTest.java @@ -54,11 +54,11 @@ void withRetryAndBackpressureAndSequentialAreCaptured() { .withBackpressure(2_000, 1_000) .withProcessingMode(ProcessingMode.SEQUENTIAL); - assertEquals(5, stream.maxRetries()); - assertEquals(java.time.Duration.ofMillis(100), stream.retryBackoff()); - assertEquals(2_000L, stream.backpressureHigh()); - assertEquals(1_000L, stream.backpressureLow()); - assertEquals(ProcessingMode.SEQUENTIAL, stream.processingMode()); + assertEquals(5, stream.consumerConfig().maxRetries()); + assertEquals(java.time.Duration.ofMillis(100), stream.consumerConfig().retryBackoff()); + assertEquals(2_000L, stream.consumerConfig().backpressureHigh()); + assertEquals(1_000L, stream.consumerConfig().backpressureLow()); + assertEquals(ProcessingMode.SEQUENTIAL, stream.consumerConfig().processingMode()); } @Test @@ -67,10 +67,10 @@ void defaultBackpressureUsesStandardWatermarks() { // Stream.withBackpressure() must produce the same defaults as the BackpressureController // constants. Hardcoded literals here would drift if the constants change; assert against the // single source of truth and pin the literal values as a separate smoke check. - assertEquals(BackpressureController.DEFAULT_HIGH_WATERMARK, stream.backpressureHigh()); - assertEquals(BackpressureController.DEFAULT_LOW_WATERMARK, stream.backpressureLow()); - assertEquals(10_000L, stream.backpressureHigh()); - assertEquals(7_000L, stream.backpressureLow()); + assertEquals(BackpressureController.DEFAULT_HIGH_WATERMARK, stream.consumerConfig().backpressureHigh()); + assertEquals(BackpressureController.DEFAULT_LOW_WATERMARK, stream.consumerConfig().backpressureLow()); + assertEquals(10_000L, stream.consumerConfig().backpressureHigh()); + assertEquals(7_000L, stream.consumerConfig().backpressureLow()); } @Test @@ -302,7 +302,9 @@ void multiBuilderAcceptsConsumerWideRetryDlqBackpressureErrorHandlerPollTimeout( @Test void multiBuilderRejectsPerRouteWithMetrics() { - final var multi = KPipe.multi(props()).json("topic-a", s -> s.withMetrics(ConsumerMetrics.noop()).toCustom(_ -> {})); + final var multi = KPipe.multi(props()).json("topic-a", s -> + s.withMetrics(ConsumerMetrics.noop()).toCustom(_ -> {}) + ); final var ex = assertThrows(IllegalArgumentException.class, multi::start); assertTrue(ex.getMessage().contains("withMetrics"), () -> "message should name withMetrics: " + ex.getMessage()); assertTrue(