fix(outbox): build dedicated raw-bytes producer for the relay (#332) #19

Merged

hibryda merged 3 commits from fix/outbox-bytes-producer-332 into main 2026-05-31 17:39:40 +02:00

Conversation 3 Commits 3 Files changed 4 +180 -9

hibryda commented 2026-05-31 17:23:05 +02:00

Owner

Copy link

Header

Project: im2be-platform-libs · Module: outbox-publisher · Issue: #332 (ADR-0015 error-event stream) · Type: fix (silent-failure)

TL;DR

BLOCKED→FIXED — the outbox relay's autoconfig injected a KafkaTemplate<byte[],byte[]> but no byte[] producer existed, so Spring wired the service's String/JSON default by generic erasure → every outbox row failed with SerializationException: Can't convert key of class [B to StringSerializer, silently breaking ADR-0014 delivery (and the ADR-0015 error-event stream on top of it).

Summary

Scope: OutboxAutoConfiguration + its IT/context tests. The relay now builds its own raw-bytes producer instead of depending on an ambiently-present byte[] KafkaTemplate. Verified: full module build green, zero compile warnings (failOnWarning=true), the integration test rewired to reproduce the production wiring. Commit c5a26e3.

Findings / Changes

#	File	Change
1	outbox-publisher/.../OutboxAutoConfiguration.java	Add buildOutboxKafkaTemplate(KafkaProperties, ObjectProvider<SslBundles>) — derives props from the service's spring.kafka.* via KafkaProperties.buildProducerProperties(SslBundles), then forces key+value ByteArraySerializer, acks=all, enable.idempotence=true. outboxPublisher + outboxPollerWorker @Bean methods now take KafkaProperties+ObjectProvider<SslBundles> and call the helper, instead of injecting KafkaTemplate<byte[],byte[]>.
2	application-it.properties	Service default serializers byte[] → String + keep acks=1 — closes the test gap that masked the prod bug; the IT now exercises the exact prod wiring.
3	OutboxBackendSelectionTest.java	Supplies KafkaProperties (autoconfig no longer injects a KafkaTemplate); dropped the now-unused raw-mock @SuppressWarnings("unchecked").
4	README.md	New "Outbox relay producer" section documenting the dedicated byte[] template + why it is built inside the bean method (not a @Bean).

Why built inside the bean method (not a @Bean)

KafkaAutoConfiguration.kafkaTemplate / kafkaProducerFactory are @ConditionalOnMissingBean. Contributing a competing KafkaTemplate/ProducerFactory bean would suppress the service's default producer and break every non-outbox Kafka send. Keeping the byte[] template private to the relay leaves the service default untouched.

Why force (not putIfAbsent) acks=all + idempotence

A durable error-event relay must never silently inherit a weaker acks from a service's general-purpose producer — it owns its at-least-once guarantee. No-op for the 7 services (already acks=all + idempotent); load-bearing only against a weaker config (e.g. the IT's acks=1). Forcing both together keeps them internally consistent (idempotence requires acks=all).

Verdict

READY — fixes a silent prod data-loss path with a regression test that now reproduces it. No public API change (constructors unchanged). Gating the deploy on the operator (image rebuild of the 6 local services).

Risk indicators

• Touches Kafka producer wiring (sensitive) — mitigated: service default producer provably untouched (built-inside-bean rationale above) + IT proves independence.
• Migration touches: none. Dependency changes: none.
• Test delta: +1 regression path (IT rewired), +KafkaProperties in context test.

Footer

im2be-platform-libs • fix • #332 • files=4 (+81/-9) • build=SUCCESS warnings=0 tests=4IT+units(0 fail) • 2026-05-31

## Header **Project:** im2be-platform-libs · **Module:** outbox-publisher · **Issue:** #332 (ADR-0015 error-event stream) · **Type:** fix (silent-failure) ## TL;DR `BLOCKED→FIXED` — the outbox relay's autoconfig injected a `KafkaTemplate<byte[],byte[]>` but no byte[] producer existed, so Spring wired the service's String/JSON default by generic erasure → **every** outbox row failed with `SerializationException: Can't convert key of class [B to StringSerializer`, silently breaking ADR-0014 delivery (and the ADR-0015 error-event stream on top of it). ## Summary Scope: `OutboxAutoConfiguration` + its IT/context tests. The relay now **builds its own** raw-bytes producer instead of depending on an ambiently-present byte[] `KafkaTemplate`. Verified: full module build green, zero compile warnings (`failOnWarning=true`), the integration test rewired to reproduce the production wiring. Commit `c5a26e3`. ## Findings / Changes | # | File | Change | |---|------|--------| | 1 | `outbox-publisher/.../OutboxAutoConfiguration.java` | Add `buildOutboxKafkaTemplate(KafkaProperties, ObjectProvider<SslBundles>)` — derives props from the service's `spring.kafka.*` via `KafkaProperties.buildProducerProperties(SslBundles)`, then **forces** key+value `ByteArraySerializer`, `acks=all`, `enable.idempotence=true`. `outboxPublisher` + `outboxPollerWorker` `@Bean` methods now take `KafkaProperties`+`ObjectProvider<SslBundles>` and call the helper, instead of injecting `KafkaTemplate<byte[],byte[]>`. | | 2 | `application-it.properties` | Service default serializers `byte[]` → **String** + keep `acks=1` — closes the test gap that masked the prod bug; the IT now exercises the exact prod wiring. | | 3 | `OutboxBackendSelectionTest.java` | Supplies `KafkaProperties` (autoconfig no longer injects a `KafkaTemplate`); dropped the now-unused raw-mock `@SuppressWarnings("unchecked")`. | | 4 | `README.md` | New "Outbox relay producer" section documenting the dedicated byte[] template + why it is built inside the bean method (not a `@Bean`). | ### Why built inside the bean method (not a `@Bean`) `KafkaAutoConfiguration.kafkaTemplate` / `kafkaProducerFactory` are `@ConditionalOnMissingBean`. Contributing a competing `KafkaTemplate`/`ProducerFactory` bean would **suppress the service's default producer** and break every non-outbox Kafka send. Keeping the byte[] template private to the relay leaves the service default untouched. ### Why force (not `putIfAbsent`) acks=all + idempotence A durable error-event relay must never silently inherit a weaker `acks` from a service's general-purpose producer — it owns its at-least-once guarantee. No-op for the 7 services (already `acks=all` + idempotent); load-bearing only against a weaker config (e.g. the IT's `acks=1`). Forcing both together keeps them internally consistent (idempotence requires `acks=all`). ## Verdict `READY` — fixes a silent prod data-loss path with a regression test that now reproduces it. No public API change (constructors unchanged). Gating the deploy on the operator (image rebuild of the 6 local services). ## Risk indicators - Touches Kafka producer wiring (sensitive) — mitigated: service default producer provably untouched (built-inside-bean rationale above) + IT proves independence. - Migration touches: none. Dependency changes: none. - Test delta: +1 regression path (IT rewired), +`KafkaProperties` in context test. ## Footer im2be-platform-libs • fix • #332 • files=4 (+81/-9) • build=SUCCESS warnings=0 tests=4IT+units(0 fail) • 2026-05-31

hibryda added 1 commit 2026-05-31 17:23:06 +02:00

fix(outbox): build dedicated raw-bytes producer for the relay (#332) ... c5a26e3766

OutboxAutoConfiguration injected a KafkaTemplate<byte[],byte[]> but no byte[]
producer existed, so Spring satisfied it from the service's String/JSON default
template by generic erasure -> every outbox row failed with
"SerializationException: Can't convert key of class [B to StringSerializer",
silently breaking ADR-0014 outbox delivery (incl. the ADR-0015 error-event
stream).

Build the relay's producer internally instead: buildOutboxKafkaTemplate()
derives props from the service's spring.kafka.* config via
KafkaProperties.buildProducerProperties(SslBundles), then forces
ByteArraySerializer (key+value) + acks=all + enable.idempotence=true. The
durability pair is forced (not putIfAbsent) because a durable error-event relay
must never inherit a weaker acks from a service's general-purpose producer; it
owns its at-least-once guarantee (no-op for the 7 services, which already set
acks=all + idempotence). It is built INSIDE the outboxPublisher /
outboxPollerWorker @Bean methods rather than as a @Bean so it does not trip
KafkaAutoConfiguration's @ConditionalOnMissingBean guards and suppress the
service's default producer.

Close the test gap: OutboxPublisherIT previously ran with byte[] serializers on
the service default, masking the production failure. It now uses String
serializers + acks=1 (mirroring prod), proving the relay's template is
independent of -- and strengthens -- the service producer config.
OutboxBackendSelectionTest now supplies KafkaProperties (the autoconfig no
longer injects a KafkaTemplate).

Verification:
- mvn -pl outbox-publisher -am clean install -> BUILD SUCCESS; Tests run: 4
  (OutboxPublisherIT) + units, 0 failures/errors
- mvn -o clean compile test-compile -> 0 [WARNING] (failOnWarning=true)

pr-reviewer-bot commented 2026-05-31 17:26:48 +02:00

Member

Copy link

Superseded by round 2.

Show previous round

hib-pr-reviewer review — PR #19 (affinity-intelligence-rework/im2be-platform-libs)

Round 1 — head c5a26e376601, base main, trigger opened

TL;DR: NEEDS_WORK — kept 2 agreed findings (1 major, 1 minor), both independently confirmed by Read/Grep; no prior run history; first arbitration for this PR.

Summary

Reconciliation complete — first run for this PR.

No prior Memora run history found. Both reviewer A and reviewer B independently raised the same two findings against the same file (OutboxAutoConfiguration.java). Code inspection at HEAD (c5a26e3) fully confirmed both:

• Major (line 396): new DefaultKafkaProducerFactory<>(props) is instantiated inline, passed directly into new KafkaTemplate<>(...), and never exposed as a Spring bean. DefaultKafkaProducerFactory implements DisposableBean specifically to flush/close its internal KafkaProducer pool; because it is never managed by the application context its destroy() is never called. A grep across the entire outbox package (*.java) confirmed that neither OutboxPublisher nor OutboxPollerWorker implement DisposableBean, Closeable, or carry @PreDestroy, leaving no secondary cleanup path. Records buffered at shutdown (non-zero linger.ms) may be dropped — a direct correctness gap for an at-least-once relay.

• Minor (lines 314 and 342): buildOutboxKafkaTemplate is a private static method with no state caching. Both outboxPublisher and outboxPollerWorker (both active by default — matchIfMissing = true) each invoke it independently, producing two separate DefaultKafkaProducerFactory instances with two independent KafkaProducer thread pools and broker TCP connections, for no isolation benefit.

Both findings kept. Severity preserved as-is. Recommended fix: convert the static helper to an instance method, store the factory as a field on OutboxAutoConfiguration (which IS a Spring singleton), and add @PreDestroy to call factory.destroy() at context shutdown — this simultaneously addresses the lifecycle leak and eliminates the duplicate-factory problem.

Blast Radius

The diff touches one auto-configuration class plus its test and properties files. The affected surface is limited to the outbox relay submodule, but OutboxAutoConfiguration is the sole wiring point for all relay beans consumed by 7 downstream services, making the lifecycle leak a production-correctness concern on any service that hosts the outbox.

BLAST_SCORE: 4/10

Risk Indicators

Indicator	Value
Sensitive functions	buildOutboxKafkaTemplate, OutboxPublisher, OutboxPollerWorker
Migration touched	—
Test delta	—
Dependency changes	—

CI status (head c5a26e376601)

Overall: ✗ failure

2 checks: 2 pending

Check	State	Link
im2be-platform-libs CI / mvn install (pull_request)	⏳ pending	details
im2be-platform-libs CI / mvn verify (main only) (pull_request)	⏳ pending	details

Findings (2)

[MAJOR] DefaultKafkaProducerFactory is never destroyed — producer threads and connections leak on shutdown

outbox-publisher/src/main/java/com/aim2be/platform/outbox/OutboxAutoConfiguration.java:396

Line 396: return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));

The DefaultKafkaProducerFactory is constructed inline and never held by any Spring-managed bean, so Spring's DisposableBean contract is never invoked on it at context shutdown. DefaultKafkaProducerFactory implements DisposableBean specifically to flush and close its internal KafkaProducer pool; skipping destroy() means:

1. The JVM-level sender thread, metadata-fetch cycle, and all broker TCP connections are abandoned rather than gracefully closed.
2. Records buffered in the producer's internal send buffer (when linger.ms > 0) are silently dropped at JVM exit — a correctness gap for an at-least-once relay during rolling deploys.
3. OutboxPublisher and OutboxPollerWorker do not implement DisposableBean, Closeable, or carry @PreDestroy (confirmed by grep across the outbox package), so there is no secondary cleanup path.

Fix: convert buildOutboxKafkaTemplate from private static to a non-static instance method on OutboxAutoConfiguration (which is itself a Spring-managed singleton), cache the factory as an instance field, and add @PreDestroy:

private DefaultKafkaProducerFactory<byte[], byte[]> outboxProducerFactory;

@PreDestroy
void destroyOutboxProducerFactory() {
    if (outboxProducerFactory != null) outboxProducerFactory.destroy();
}

private KafkaTemplate<byte[], byte[]> buildOutboxKafkaTemplate(
        KafkaProperties kafkaProperties, ObjectProvider<SslBundles> sslBundles) {
    if (outboxProducerFactory == null) {
        final Map<String, Object> props =
                kafkaProperties.buildProducerProperties(sslBundles.getIfAvailable());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        outboxProducerFactory = new DefaultKafkaProducerFactory<>(props);
    }
    return new KafkaTemplate<>(outboxProducerFactory);
}

This also resolves the minor duplicate-factory finding.

[MINOR] buildOutboxKafkaTemplate called twice — two independent producer factories open doubled broker connections

outbox-publisher/src/main/java/com/aim2be/platform/outbox/OutboxAutoConfiguration.java:314

Lines 314 and 342 each call buildOutboxKafkaTemplate(kafkaProperties, sslBundles) independently. The method is private static with no caching, so each invocation allocates a fresh DefaultKafkaProducerFactory. Both outboxPublisher and outboxPollerWorker are active by default (matchIfMissing = true), meaning at runtime the relay holds two separate KafkaProducer instances with independent sender threads, metadata-fetch cycles, and TCP connections to the Kafka broker — for no isolation benefit, as both beans form a single logical relay.

Resolved as a side-effect of the lifecycle-leak fix above: caching the factory as an instance field means both bean methods share one DefaultKafkaProducerFactory and one broker connection pool.

Verdict

NEEDS_WORK

---

_{hib-pr-reviewer • round 1 • 2 findings (1M/1m) • 2026-05-31T15:25:32.968Z → 2026-05-31T15:26:48.138Z • posted-as: pr-reviewer-bot • model: auto}

<!-- hib-pr-reviewer collapsed --> > _Superseded by round 2._ <details> <summary>Show previous round</summary> <!-- hib-pr-reviewer round:1 --> ## hib-pr-reviewer review — PR #19 (affinity-intelligence-rework/im2be-platform-libs) **Round 1** — head `c5a26e376601`, base `main`, trigger `opened` **TL;DR:** NEEDS_WORK — kept 2 agreed findings (1 major, 1 minor), both independently confirmed by Read/Grep; no prior run history; first arbitration for this PR. ### Summary **Reconciliation complete — first run for this PR.** No prior Memora run history found. Both reviewer A and reviewer B independently raised the same two findings against the same file (`OutboxAutoConfiguration.java`). Code inspection at HEAD (`c5a26e3`) fully confirmed both: - **Major (line 396)**: `new DefaultKafkaProducerFactory<>(props)` is instantiated inline, passed directly into `new KafkaTemplate<>(...)`, and never exposed as a Spring bean. `DefaultKafkaProducerFactory` implements `DisposableBean` specifically to flush/close its internal `KafkaProducer` pool; because it is never managed by the application context its `destroy()` is never called. A `grep` across the entire outbox package (`*.java`) confirmed that neither `OutboxPublisher` nor `OutboxPollerWorker` implement `DisposableBean`, `Closeable`, or carry `@PreDestroy`, leaving no secondary cleanup path. Records buffered at shutdown (non-zero `linger.ms`) may be dropped — a direct correctness gap for an at-least-once relay. - **Minor (lines 314 and 342)**: `buildOutboxKafkaTemplate` is a `private static` method with no state caching. Both `outboxPublisher` and `outboxPollerWorker` (both active by default — `matchIfMissing = true`) each invoke it independently, producing two separate `DefaultKafkaProducerFactory` instances with two independent `KafkaProducer` thread pools and broker TCP connections, for no isolation benefit. Both findings kept. Severity preserved as-is. Recommended fix: convert the static helper to an instance method, store the factory as a field on `OutboxAutoConfiguration` (which IS a Spring singleton), and add `@PreDestroy` to call `factory.destroy()` at context shutdown — this simultaneously addresses the lifecycle leak and eliminates the duplicate-factory problem. ### Blast Radius The diff touches one auto-configuration class plus its test and properties files. The affected surface is limited to the outbox relay submodule, but `OutboxAutoConfiguration` is the sole wiring point for all relay beans consumed by 7 downstream services, making the lifecycle leak a production-correctness concern on any service that hosts the outbox. **BLAST_SCORE: 4/10** ### Risk Indicators | Indicator | Value | |---|---| | Sensitive functions | `buildOutboxKafkaTemplate`, `OutboxPublisher`, `OutboxPollerWorker` | | Migration touched | — | | Test delta | — | | Dependency changes | — | ### CI status (head `c5a26e376601`) **Overall: ✗ failure** 2 checks: 2 pending | Check | State | Link | |---|---|---| | im2be-platform-libs CI / mvn install (pull_request) | ⏳ pending | [details](/affinity-intelligence-rework/im2be-platform-libs/actions/runs/105/jobs/0) | | im2be-platform-libs CI / mvn verify (main only) (pull_request) | ⏳ pending | [details](/affinity-intelligence-rework/im2be-platform-libs/actions/runs/105/jobs/1) | ### Findings (2) #### **[MAJOR]** DefaultKafkaProducerFactory is never destroyed — producer threads and connections leak on shutdown _outbox-publisher/src/main/java/com/aim2be/platform/outbox/OutboxAutoConfiguration.java:396_ Line 396: `return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));` The `DefaultKafkaProducerFactory` is constructed inline and never held by any Spring-managed bean, so Spring's `DisposableBean` contract is never invoked on it at context shutdown. `DefaultKafkaProducerFactory` implements `DisposableBean` specifically to flush and close its internal `KafkaProducer` pool; skipping `destroy()` means: 1. The JVM-level sender thread, metadata-fetch cycle, and all broker TCP connections are abandoned rather than gracefully closed. 2. Records buffered in the producer's internal send buffer (when `linger.ms > 0`) are silently dropped at JVM exit — a correctness gap for an at-least-once relay during rolling deploys. 3. `OutboxPublisher` and `OutboxPollerWorker` do not implement `DisposableBean`, `Closeable`, or carry `@PreDestroy` (confirmed by grep across the outbox package), so there is no secondary cleanup path. **Fix**: convert `buildOutboxKafkaTemplate` from `private static` to a non-static instance method on `OutboxAutoConfiguration` (which is itself a Spring-managed singleton), cache the factory as an instance field, and add `@PreDestroy`: ```java private DefaultKafkaProducerFactory<byte[], byte[]> outboxProducerFactory; @PreDestroy void destroyOutboxProducerFactory() { if (outboxProducerFactory != null) outboxProducerFactory.destroy(); } private KafkaTemplate<byte[], byte[]> buildOutboxKafkaTemplate( KafkaProperties kafkaProperties, ObjectProvider<SslBundles> sslBundles) { if (outboxProducerFactory == null) { final Map<String, Object> props = kafkaProperties.buildProducerProperties(sslBundles.getIfAvailable()); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); props.put(ProducerConfig.ACKS_CONFIG, "all"); props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); outboxProducerFactory = new DefaultKafkaProducerFactory<>(props); } return new KafkaTemplate<>(outboxProducerFactory); } ``` This also resolves the minor duplicate-factory finding. #### **[MINOR]** buildOutboxKafkaTemplate called twice — two independent producer factories open doubled broker connections _outbox-publisher/src/main/java/com/aim2be/platform/outbox/OutboxAutoConfiguration.java:314_ Lines 314 and 342 each call `buildOutboxKafkaTemplate(kafkaProperties, sslBundles)` independently. The method is `private static` with no caching, so each invocation allocates a fresh `DefaultKafkaProducerFactory`. Both `outboxPublisher` and `outboxPollerWorker` are active by default (`matchIfMissing = true`), meaning at runtime the relay holds two separate `KafkaProducer` instances with independent sender threads, metadata-fetch cycles, and TCP connections to the Kafka broker — for no isolation benefit, as both beans form a single logical relay. Resolved as a side-effect of the lifecycle-leak fix above: caching the factory as an instance field means both bean methods share one `DefaultKafkaProducerFactory` and one broker connection pool. ### Verdict **NEEDS_WORK** --- _{hib-pr-reviewer • round 1 • 2 findings (1M/1m) • 2026-05-31T15:25:32.968Z → 2026-05-31T15:26:48.138Z • posted-as: pr-reviewer-bot • model: auto} </details>

hibryda added 1 commit 2026-05-31 17:30:17 +02:00

fix(outbox): manage relay producer factory lifecycle (#332) ... 1b60f3fe6c

R1 verdict NEEDS_WORK findings (kept=2):

(1) MAJOR OutboxAutoConfiguration.java:396 — DefaultKafkaProducerFactory never
destroyed. The factory was instantiated inline inside buildOutboxKafkaTemplate
and never held by a Spring-managed bean, so its DisposableBean.destroy() never
fired at context shutdown -> leaked the sender thread + broker TCP connections
and risked dropping linger.ms-buffered records (an at-least-once gap on rolling
deploys).

(2) MINOR OutboxAutoConfiguration.java:314,342 — buildOutboxKafkaTemplate
(private static, uncached) was called once per relay bean (outboxPublisher +
outboxPollerWorker, both active by default) -> two independent factories /
connection pools for one logical relay.

Fix: cache the DefaultKafkaProducerFactory on the auto-configuration singleton
(outboxProducerFactory field, lazy-built in the now-instance
buildOutboxKafkaTemplate — safe unsynchronized: bean methods run only during
single-threaded context refresh), shared by both bean methods (one connection
pool), and add @PreDestroy closeOutboxProducerFactory() calling
factory.destroy() for graceful flush+close. Resolves both findings together.

Verification:
- mvn -pl outbox-publisher -am clean install -> BUILD SUCCESS; 103 unit + 4 IT,
  0 failures/errors
- 0 [WARNING] (failOnWarning=true)
- ripple: the prior IT 30s "Surefire is going to kill self fork JVM" shutdown
  delay is GONE — the producer now closes cleanly at @PreDestroy.

pr-reviewer-bot commented 2026-05-31 17:34:12 +02:00

Member

Copy link

Superseded by round 3.

Show previous round

hib-pr-reviewer review — PR #19 (affinity-intelligence-rework/im2be-platform-libs)

Round 2 — head 1b60f3fe6cce, base main, trigger synchronize

TL;DR: CONDITIONAL_APPROVE — kept 1 unique-to-A minor finding (verified at line 434: missing MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION cap under forced idempotence); both round-1 blocking findings confirmed resolved.

Summary

Arbitration summary — Round 2

Prior-round findings: Both round-1 findings (producer-factory resource leak, duplicate-factory construction) confirmed resolved by both reviewers. Not re-raised.

Unique-to-A (1 finding): Reviewer A flagged that buildOutboxKafkaTemplate forces enable.idempotence=true (line 434) but does not also cap max.in.flight.requests.per.connection to ≤ 5. Verified via Read of lines 425–438: the four forced knobs are KEY_SERIALIZER_CLASS_CONFIG, VALUE_SERIALIZER_CLASS_CONFIG, ACKS_CONFIG, and ENABLE_IDEMPOTENCE_CONFIG — MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION is absent. A service-level override above 5 carried in via buildProducerProperties() would survive unmodified and cause ConfigException at the first lazy-producer send. Finding is kept as minor.

Unique-to-B: Reviewer B raised no new findings. Nothing to evaluate.

Net result: 1 agreed-or-kept finding, 0 dropped, verdict CONDITIONAL_APPROVE.

Blast Radius

Changes are tightly scoped to OutboxAutoConfiguration (a single auto-configuration class), its companion test, and README documentation. The new buildOutboxKafkaTemplate helper is private and called only during single-threaded context refresh; no exported surface is altered. The blast radius is low: only services that import outbox-publisher are affected, and the behavioral change (dedicated byte[] producer) is isolated from the service's default KafkaTemplate.

BLAST_SCORE: 2/10

Risk Indicators

Indicator	Value
Sensitive functions	buildOutboxKafkaTemplate, closeOutboxProducerFactory
Migration touched	—
Test delta	—
Dependency changes	—

CI status (head 1b60f3fe6cce)

Overall: ✗ failure

2 checks: 2 pending

Check	State	Link
im2be-platform-libs CI / mvn install (pull_request)	⏳ pending	details
im2be-platform-libs CI / mvn verify (main only) (pull_request)	⏳ pending	details

Findings (1)

[MINOR] Forcing enable.idempotence=true without capping max.in.flight.requests.per.connection ≤ 5 can produce a silent ConfigException at first send

outbox-publisher/src/main/java/com/aim2be/platform/outbox/OutboxAutoConfiguration.java:434

Lines 431–434 force four producer knobs (ByteArraySerializer × 2, acks=all, enable.idempotence=true), but omit MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION. The props map is first populated by kafkaProperties.buildProducerProperties(sslBundles.getIfAvailable()) (line 430), which faithfully copies every spring.kafka.producer.properties.* override the service has declared. If a future adopter sets spring.kafka.producer.properties.max.in.flight.requests.per.connection above 5 for throughput (the Kafka default is 5, so this requires a deliberate override), that value will be present in props and will not be overwritten. Kafka enforces the ≤ 5 constraint eagerly during KafkaProducer construction; because DefaultKafkaProducerFactory is lazy, this ConfigException surfaces only at the first enqueue attempt — not at application startup — making it hard to diagnose.

The 7 existing services are unaffected (they already configure idempotence and thus already comply), but the risk is structural: every other forced durability knob here has a corresponding explicit override; this one does not.

Fix: add props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5) alongside the other forced knobs. 5 is Kafka's maximum permissible value under idempotence and preserves ordering guarantees consistent with the relay's at-least-once contract. Alternatively, use props.merge(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5, (existing, cap) -> Math.min((Integer) existing, (Integer) cap)) if preserving compliant values already set by the service is desirable — but the simpler forced-5 approach is consistent with how acks and idempotence are handled here.

Verdict

CONDITIONAL_APPROVE

---

_{hib-pr-reviewer • round 2 • 1 finding (1m) • 2026-05-31T15:32:46.264Z → 2026-05-31T15:34:12.020Z • posted-as: pr-reviewer-bot • model: auto}

<!-- hib-pr-reviewer collapsed --> > _Superseded by round 3._ <details> <summary>Show previous round</summary> <!-- hib-pr-reviewer round:2 --> ## hib-pr-reviewer review — PR #19 (affinity-intelligence-rework/im2be-platform-libs) **Round 2** — head `1b60f3fe6cce`, base `main`, trigger `synchronize` **TL;DR:** CONDITIONAL_APPROVE — kept 1 unique-to-A minor finding (verified at line 434: missing MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION cap under forced idempotence); both round-1 blocking findings confirmed resolved. ### Summary ## Arbitration summary — Round 2 **Prior-round findings:** Both round-1 findings (producer-factory resource leak, duplicate-factory construction) confirmed resolved by both reviewers. Not re-raised. **Unique-to-A (1 finding):** Reviewer A flagged that `buildOutboxKafkaTemplate` forces `enable.idempotence=true` (line 434) but does not also cap `max.in.flight.requests.per.connection` to ≤ 5. Verified via `Read` of lines 425–438: the four forced knobs are `KEY_SERIALIZER_CLASS_CONFIG`, `VALUE_SERIALIZER_CLASS_CONFIG`, `ACKS_CONFIG`, and `ENABLE_IDEMPOTENCE_CONFIG` — `MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION` is absent. A service-level override above 5 carried in via `buildProducerProperties()` would survive unmodified and cause `ConfigException` at the first lazy-producer send. Finding is **kept** as minor. **Unique-to-B:** Reviewer B raised no new findings. Nothing to evaluate. **Net result:** 1 agreed-or-kept finding, 0 dropped, verdict CONDITIONAL_APPROVE. ### Blast Radius Changes are tightly scoped to `OutboxAutoConfiguration` (a single auto-configuration class), its companion test, and README documentation. The new `buildOutboxKafkaTemplate` helper is private and called only during single-threaded context refresh; no exported surface is altered. The blast radius is low: only services that import `outbox-publisher` are affected, and the behavioral change (dedicated byte[] producer) is isolated from the service's default `KafkaTemplate`. **BLAST_SCORE: 2/10** ### Risk Indicators | Indicator | Value | |---|---| | Sensitive functions | `buildOutboxKafkaTemplate`, `closeOutboxProducerFactory` | | Migration touched | — | | Test delta | — | | Dependency changes | — | ### CI status (head `1b60f3fe6cce`) **Overall: ✗ failure** 2 checks: 2 pending | Check | State | Link | |---|---|---| | im2be-platform-libs CI / mvn install (pull_request) | ⏳ pending | [details](/affinity-intelligence-rework/im2be-platform-libs/actions/runs/106/jobs/0) | | im2be-platform-libs CI / mvn verify (main only) (pull_request) | ⏳ pending | [details](/affinity-intelligence-rework/im2be-platform-libs/actions/runs/106/jobs/1) | ### Findings (1) #### **[MINOR]** Forcing `enable.idempotence=true` without capping `max.in.flight.requests.per.connection` ≤ 5 can produce a silent `ConfigException` at first send _outbox-publisher/src/main/java/com/aim2be/platform/outbox/OutboxAutoConfiguration.java:434_ Lines 431–434 force four producer knobs (`ByteArraySerializer` × 2, `acks=all`, `enable.idempotence=true`), but omit `MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION`. The props map is first populated by `kafkaProperties.buildProducerProperties(sslBundles.getIfAvailable())` (line 430), which faithfully copies every `spring.kafka.producer.properties.*` override the service has declared. If a future adopter sets `spring.kafka.producer.properties.max.in.flight.requests.per.connection` above 5 for throughput (the Kafka default is 5, so this requires a deliberate override), that value will be present in `props` and will not be overwritten. Kafka enforces the ≤ 5 constraint eagerly during `KafkaProducer` construction; because `DefaultKafkaProducerFactory` is lazy, this `ConfigException` surfaces only at the first enqueue attempt — not at application startup — making it hard to diagnose. The 7 existing services are unaffected (they already configure idempotence and thus already comply), but the risk is structural: every other forced durability knob here has a corresponding explicit override; this one does not. **Fix:** add `props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5)` alongside the other forced knobs. 5 is Kafka's maximum permissible value under idempotence and preserves ordering guarantees consistent with the relay's at-least-once contract. Alternatively, use `props.merge(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5, (existing, cap) -> Math.min((Integer) existing, (Integer) cap))` if preserving compliant values already set by the service is desirable — but the simpler forced-5 approach is consistent with how `acks` and `idempotence` are handled here. ### Verdict **CONDITIONAL_APPROVE** --- _{hib-pr-reviewer • round 2 • 1 finding (1m) • 2026-05-31T15:32:46.264Z → 2026-05-31T15:34:12.020Z • posted-as: pr-reviewer-bot • model: auto} </details>

hibryda added 1 commit 2026-05-31 17:35:49 +02:00

fix(outbox): cap max.in.flight=5 for the forced-idempotent relay producer (#332) ... a3e09ff0a9

R2 verdict CONDITIONAL_APPROVE findings (kept=1):

(1) MINOR OutboxAutoConfiguration.java:434 — forced enable.idempotence=true
without pinning max.in.flight.requests.per.connection. The idempotent producer
requires max.in.flight <= 5; if a service set
spring.kafka.producer.properties.max.in.flight.requests.per.connection higher
(for its own non-outbox throughput), buildProducerProperties would carry that
value into the outbox template and DefaultKafkaProducerFactory would throw
ConfigException at first send. Since we already FORCE idempotence, the relay
must own the cap it entails.

Fix: props.put(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5) alongside the forced
acks=all + idempotence. 5 is Kafka's idempotence-compatible maximum, so it
preserves both ordering and throughput. Javadoc + README updated to "five
forced knobs" with the mutual-consistency rationale (idempotence requires
acks=all AND max.in.flight<=5).

R1's two blocking findings (factory lifecycle leak + duplicate factory) were
confirmed RESOLVED by R2.

Verification:
- mvn -pl outbox-publisher -am clean install -> BUILD SUCCESS; 103 unit + 4 IT,
  0 failures/errors
- 0 [WARNING] (failOnWarning=true)

pr-reviewer-bot commented 2026-05-31 17:38:31 +02:00

Member

Copy link

hib-pr-reviewer review — PR #19 (affinity-intelligence-rework/im2be-platform-libs)

Round 3 — head a3e09ff0a9db, base main, trigger synchronize

TL;DR: NO_NEW_FINDINGS — No new findings this round.

Summary

[quorum-converged] A=0 = B=0. All three prior-round findings are resolved. The full file confirms:

1. Round-1 factory-destroy leak → closed. @PreDestroy closeOutboxProducerFactory() (lines 165–170) guards against null and calls outboxProducerFactory.destroy(), which flushes buffered records and releases the sender thread + broker TCP connections at context shutdown.

2. Round-1 doubled connections → closed. buildOutboxKafkaTemplate (lines 427–448) gates factory creation behind if (outboxProducerFactory == null) and caches the result in the field. Both outboxPublisher (line 345) and outboxPollerWorker (line 373) receive distinct KafkaTemplate wrapper instances, but they share a single underlying DefaultKafkaProducerFactory and therefore a single broker connection pool.

3. Round-2 idempotence ConfigException risk → closed. Line 444 now pins MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = 5 (Kafka's idempotence-compatible maximum), consistent with the forced enable.idempotence=true and acks=all. Kafka's AbstractConfig correctly handles the Integer and Boolean literal types used for these overrides.

Test coverage is sound: OutboxBackendSelectionTest replaces the KafkaTemplate mock with a real KafkaProperties bean (KafkaProperties::new) — construction of DefaultKafkaProducerFactory is deferred to first send, so no broker is required during the gate tests. application-it.properties now deliberately configures String serializers + acks=1 (mirroring the 7 production services), proving that the relay's own buildOutboxKafkaTemplate overrides the service-default config and that the pre-fix masking (byte-array IT defaults hiding the SerializationException) cannot recur.

No new actionable findings.

CI status (head a3e09ff0a9db)

Overall: ✗ failure

2 checks: 2 pending

Check	State	Link
im2be-platform-libs CI / mvn install (pull_request)	⏳ pending	details
im2be-platform-libs CI / mvn verify (main only) (pull_request)	⏳ pending	details

Findings

No new findings this round.

Quorum converged on empty findings (A + B both returned 0).

Verdict

NO_NEW_FINDINGS

---

_{hib-pr-reviewer • round 3 • 0 findings • 2026-05-31T15:35:50.957Z → 2026-05-31T15:38:31.231Z • posted-as: pr-reviewer-bot • model: auto • [bookkeeping fallback]}

<!-- hib-pr-reviewer round:3 --> ## hib-pr-reviewer review — PR #19 (affinity-intelligence-rework/im2be-platform-libs) **Round 3** — head `a3e09ff0a9db`, base `main`, trigger `synchronize` **TL;DR:** NO_NEW_FINDINGS — No new findings this round. ### Summary [quorum-converged] A=0 = B=0. **All three prior-round findings are resolved.** The full file confirms: 1. **Round-1 factory-destroy leak** → closed. `@PreDestroy closeOutboxProducerFactory()` (lines 165–170) guards against null and calls `outboxProducerFactory.destroy()`, which flushes buffered records and releases the sender thread + broker TCP connections at context shutdown. 2. **Round-1 doubled connections** → closed. `buildOutboxKafkaTemplate` (lines 427–448) gates factory creation behind `if (outboxProducerFactory == null)` and caches the result in the field. Both `outboxPublisher` (line 345) and `outboxPollerWorker` (line 373) receive distinct `KafkaTemplate` wrapper instances, but they share a single underlying `DefaultKafkaProducerFactory` and therefore a single broker connection pool. 3. **Round-2 idempotence `ConfigException` risk** → closed. Line 444 now pins `MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = 5` (Kafka's idempotence-compatible maximum), consistent with the forced `enable.idempotence=true` and `acks=all`. Kafka's `AbstractConfig` correctly handles the `Integer` and `Boolean` literal types used for these overrides. Test coverage is sound: `OutboxBackendSelectionTest` replaces the `KafkaTemplate` mock with a real `KafkaProperties` bean (`KafkaProperties::new`) — construction of `DefaultKafkaProducerFactory` is deferred to first send, so no broker is required during the gate tests. `application-it.properties` now deliberately configures String serializers + `acks=1` (mirroring the 7 production services), proving that the relay's own `buildOutboxKafkaTemplate` overrides the service-default config and that the pre-fix masking (byte-array IT defaults hiding the `SerializationException`) cannot recur. No new actionable findings. ### CI status (head `a3e09ff0a9db`) **Overall: ✗ failure** 2 checks: 2 pending | Check | State | Link | |---|---|---| | im2be-platform-libs CI / mvn install (pull_request) | ⏳ pending | [details](/affinity-intelligence-rework/im2be-platform-libs/actions/runs/107/jobs/0) | | im2be-platform-libs CI / mvn verify (main only) (pull_request) | ⏳ pending | [details](/affinity-intelligence-rework/im2be-platform-libs/actions/runs/107/jobs/1) | ### Findings **No new findings this round.** _Quorum converged on empty findings (A + B both returned 0)._ ### Verdict **NO_NEW_FINDINGS** --- _{hib-pr-reviewer • round 3 • 0 findings • 2026-05-31T15:35:50.957Z → 2026-05-31T15:38:31.231Z • posted-as: pr-reviewer-bot • model: auto • [bookkeeping fallback]}

hibryda merged commit a920914843 into main 2026-05-31 17:39:40 +02:00

hibryda deleted branch fix/outbox-bytes-producer-332 2026-05-31 17:39:41 +02:00

hibryda referenced this pull request from a commit 2026-05-31 17:39:41 +02:00

fix(outbox): dedicated raw-bytes producer for the relay (#332) (#19)

hibryda referenced this pull request 2026-05-31 18:34:09 +02:00

CI: mvn install runs Testcontainers IT but the runner job-container has no Docker → chronic red CI #20

Sign in to join this conversation.

Reviewers

No reviewers

Labels

Clear labels

No items

No labels

Milestone

Clear milestone

No items

No milestone

Projects

Clear projects

No items

No project

Assignees

Clear assignees

No assignees

2 participants

Notifications

Subscribe

Due date

The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set.

Reference

affinity-intelligence-rework/im2be-platform-libs!19

No description provided.