feat: add Kafka publish for build/batch state-change events by ashwgit · Pull Request #1381 · release-engineering/iib

ashwgit · 2026-09-04T12:29:27Z

Related Jira Ticket : CLOUDDST-32828

Overview
This PR introduces Kafka messaging features, allowing IIB to broadcast build and batch state-change events to Kafka topics. Implemented code make things modular so that the legacy AMQP/UMB feature could be depricated easily. This MR will :

Architectural & Security Highlights

-  Integration: The Kafka producer is designed to be completely fault-tolerant. Message dispatching is wrapped in broad exception handling, guaranteeing that Kafka connection drops or serialization errors will never bubble up to cause API errors or disrupt core request flows.

- Security: The producer structurally prevents accidental unencrypted (fail-open) connections. It strictly enforces SASL_SSL transport and SCRAM-SHA-512 authentication.

- Thread-Safe Lifecycle: Introduces a cached, thread-safe Kafka singleton (kafka_producer.py) that efficiently reuses connections across application contexts.

Configuration :

The feature is fully opt-in and driven by new IIB_KAFKA_* configuration variables (brokers, credentials, topics, and CA file paths). If IIB_KAFKA_BROKERS is omitted, the Kafka publishing path gracefully disables itself with zero application overhead.

@release-engineering/exd-guild-hello-operator PTAL

fullsend-ai-review · 2026-09-04T12:30:40Z

🤖 Finished Review · ❌ Failure (validation failed after 2 iteration(s)) · Started 12:30 PM UTC · Completed 1:12 PM UTC

Commit: 6d3a2c4 · View workflow run →

Runtime: claude · Model: opus → claude-opus-4-6 · Effort: high · Cost: $4.61

fullsend-ai-review · 2026-09-04T13:57:19Z

🤖 Finished Review · ❌ Failure (validation failed after 2 iteration(s)) · Started 1:57 PM UTC · Completed 2:39 PM UTC

Commit: fb219fb · View workflow run →

Runtime: claude · Model: opus → claude-opus-4-6 · Effort: high

fullsend-ai-review · 2026-09-04T14:40:32Z

🤖 Finished Review · ✅ Success · Started 2:40 PM UTC · Completed 3:01 PM UTC

Commit: 7dfa904 · View workflow run →

Runtime: claude · Model: opus → claude-opus-4-6 · Effort: high · Cost: $6.82

fullsend-ai-review · 2026-09-04T15:01:11Z

Risk Assessment: moderate (2/5)

Details

Large blast radius and dual dependency file changes are the primary risk drivers, but clean recent git history (no churn, single-author files, no reverts on core modules) and adequate test coverage for the new Kafka producer keep the composite score at moderate.

Previous run

Risk Assessment: elevated (3/5)

Details

Elevated risk: BLAST_RADIUS=large drives change-size to 5 and dual dependency-file changes also score 5, test coverage slipped to 0.20, and git history on messaging.py shows a pattern of hotfixes and reverts — together these push the score one point above the prior moderate assessment despite no protected paths or security-sensitive files being touched.

Previous run (2)

Risk Assessment: moderate (2/5)

Details

Moderate risk: the PR adds a new Kafka messaging transport alongside existing AMQP with a new dependency and two new modules, but risk is tempered by solid test coverage (0.33 ratio), stable git history on modified files, no protected or security-sensitive paths touched, and a non-first-time author.

fullsend-ai-review · 2026-09-04T15:01:13Z

Review

Findings

Medium

[error-handling] iib/web/messaging.py:271 — Bare except: in _send_kafka_messages catches BaseException including SystemExit and KeyboardInterrupt, which can prevent clean process shutdown. The PR's own kafka_producer.py correctly uses except Exception:, creating an internal inconsistency.
Remediation: Replace except: # noqa: E722 with except Exception:.
[resource-lifecycle] iib/web/kafka_producer.py:59 — The module-level _kafka_producer singleton is never explicitly flushed or closed. With linger_ms=5, buffered messages can be lost on process shutdown. No Flask teardown hook or signal handler calls producer.flush() / producer.close().
Remediation: Register a Flask teardown_appcontext or atexit handler that calls _kafka_producer.flush() and _kafka_producer.close().
[missing-documentation] docs/gettingstarted.md:245 — The configuration section documents AMQP 1.0 messaging options but is missing the six new Kafka configuration keys (IIB_KAFKA_BROKERS, IIB_KAFKA_USERNAME, IIB_KAFKA_PASSWORD, IIB_KAFKA_SSL_CAFILE, IIB_KAFKA_BUILD_STATE_TOPIC, IIB_KAFKA_BATCH_STATE_TOPIC). The Messaging section at line 511 still refers only to AMQP without mentioning Kafka support.
Remediation: Add Kafka configuration documentation to gettingstarted.md mirroring the block added to README.md, and update the Messaging section.

Low

[architectural-coherence] iib/web/kafka_producer.py:14 — The Kafka path introduces a process-wide module-level singleton while the existing AMQP path creates a fresh connection per call. Tests work around this by directly resetting kafka_producer._kafka_producer = None. The singleton pattern is idiomatic for KafkaProducer but diverges from the project's existing connection model.
Remediation: Document the rationale for the singleton lifetime in a module-level docstring, or scope the producer to the Flask application context.
[edge-case/authentication] iib/web/kafka_producer.py:23 — Double-checked locking caches the producer indefinitely. No reconnection, cache-invalidation, or credential-rotation mechanism exists. The application must be fully restarted for new credentials or to recover from certain failure modes.
Remediation: Add a mechanism to invalidate the cached producer (e.g., TTL-based refresh, a reset function, or a signal handler).
[secrets-handling] iib/web/kafka_producer.py:49 — Kafka password placed in producer_kwargs dict could be exposed if logger.exception() captures local variables in certain logging or APM configurations.
Remediation: Clear password from producer_kwargs in a finally block, or pass credentials as direct keyword arguments.
[TLS-configuration] iib/web/kafka_producer.py:55 — When IIB_KAFKA_SSL_CAFILE is explicitly set to a falsy value, ssl_cafile is omitted. TLS is still enforced via SASL_SSL, but the set of trusted CAs changes to the system default.
[code-organization] iib/web/kafka_producer.py:70 — import logging inside on_send_error function body diverges from the module-scope import convention. The function-level import is intentional (background thread without Flask context) but inconsistent with codebase style.
Remediation: Move import logging to the top of the module.
[error-handling-idioms] iib/web/kafka_producer.py:87 — Logger calls use f-strings; the established convention in messaging.py is %-style lazy formatting which avoids eager interpolation when the log level is suppressed.
Remediation: Replace f-strings with %-style formatting in logger calls.
[documentation-comment-format] iib/web/kafka_producer.py:81 — send_kafka_message and on_send_error lack :param:/:rtype: docstring entries used by other functions in this module and messaging.py.
Remediation: Expand docstrings to include :param: and :rtype: entries matching the style of messaging.py helpers.
[api-shape-patterns] iib/web/kafka_producer.py:68 — on_send_error(excp) has no type annotation; all other callable signatures carry explicit annotations.
Remediation: Annotate: def on_send_error(excp: Exception) -> None:.
[api-shape-patterns] iib/web/messaging.py:29 — _build_request_state_change_data and _build_batch_state_change_data return bare tuple instead of explicit Tuple[...] generics used elsewhere in the codebase.
Remediation: Use Tuple[BaseClassRequestResponse, Dict[str, Any]] and Optional[Tuple[...]].
[naming-trajectory] iib/web/config.py:41 — Existing messaging config uses IIB_MESSAGING_* prefix; this PR introduces a parallel IIB_KAFKA_* namespace. A separate namespace is reasonable given the different transport backends, but diverges from the single-prefix pattern.
Remediation: Consider IIB_MESSAGING_KAFKA_* to keep messaging config under one prefix, or document the naming rationale.
[architectural-coherence] iib/web/config.py:58 — DevelopmentConfig pre-populates sample AMQP values for developer discoverability but has no equivalent Kafka sample values.
Remediation: Add commented-out sample Kafka config values to DevelopmentConfig.
[test-adequacy] tests/test_web/test_messaging.py:312 — test_send_messages_for_new_batch_of_requests_no_requests mocks _send_kafka_messages but does not assert it is NOT called when the request list is empty.
Remediation: Add mock_skm.assert_not_called().
[scope-creep] README.md:526 — States AMQP 'will be deprecated' but the PR provides no deprecation warning, migration guide, or timeline.
Remediation: Remove the deprecation claim from the README or open a follow-up issue tracking AMQP deprecation.
[documentation-quality] README.md:526 — Contains 'depricated' (typo for 'deprecated') and 'Please notethat' (missing space between 'note' and 'that').
Remediation: Correct to 'deprecated' and 'Please note that'.

Next steps:

/fs-fix — agent addresses review findings automatically
/fs-fix <your instruction> — agent fixes with your specific guidance
Push commits directly — review re-runs automatically on push
/fs-fix-stop — disable automatic fix runs for this PR

Previous run

Review

Reason: stale-head

The review agent reviewed commit 45e62d035aa709d6e479c96051dfd6fd4c667aba but the PR HEAD is now 51173ac5bb449e552dcc85fed317fdc65b075e87. This review was discarded to avoid approving unreviewed code.

Previous run (2)

Review

Reason: stale-head

The review agent reviewed commit 7dfa904eee0d2205692653d1427992605b8fb75b but the PR HEAD is now dce85fc8259a4dc835b81221959627ba33562db4. This review was discarded to avoid approving unreviewed code.

fullsend-ai-review · 2026-09-04T15:01:15Z

/fs-review

fullsend-ai-review · 2026-09-04T15:02:35Z

🤖 Finished Review · ❌ Failure (validation failed after 2 iteration(s)) · Started 3:02 PM UTC · Completed 3:44 PM UTC

Commit: dce85fc · View workflow run →

Runtime: claude · Model: opus → claude-opus-4-6 · Effort: high

fullsend-ai-review · 2026-09-07T07:39:45Z

🤖 Finished Review · ✅ Success · Started 7:39 AM UTC · Completed 8:01 AM UTC

Commit: 45e62d0 · View workflow run →

Runtime: claude · Model: opus → claude-opus-4-6 · Effort: high · Cost: $6.43

Introduces Kafka messaging features, allowing IIB to broadcast build and batch state-change events to Kafka topics. - Added kafka-python support in IIB - Kafka messaging is optional and if brokers are not configured, IIB will skip trying sending messages to Kafka

fullsend-ai-review · 2026-09-07T08:01:07Z

/fs-review

fullsend-ai-review · 2026-09-07T08:02:12Z

🤖 Finished Review · ✅ Success · Started 8:02 AM UTC · Completed 8:23 AM UTC

Commit: 51173ac · View workflow run →

Runtime: claude · Model: opus → claude-opus-4-6 · Effort: high · Cost: $6.23

fullsend-ai-review

Note: The following review comments could not be posted on the diff (GitHub returned 422) and are included here instead:

iib/web/config.py (file-level): Line 58 · [low] architectural-coherence

DevelopmentConfig pre-populates sample AMQP values for developer discoverability but has no equivalent Kafka sample values.

Suggested fix: Add commented-out sample Kafka config values to DevelopmentConfig.

fullsend-ai-review

See the review comment for full details.

fullsend-ai-review · 2026-09-07T08:22:54Z

+            if batch_data:
+                content, properties = batch_data
+                send_kafka_message(producer, batch_topic, content, properties)
+    except:  # noqa: E722


[medium] error-handling

Bare except: in _send_kafka_messages catches BaseException including SystemExit and KeyboardInterrupt, which can prevent clean process shutdown. The PR's own kafka_producer.py correctly uses except Exception:, creating an internal inconsistency.

Suggested fix: Replace except: # noqa: E722 with except Exception:.

fullsend-ai-review · 2026-09-07T08:22:54Z

+            if ca_file:
+                producer_kwargs['ssl_cafile'] = ca_file
+
+            _kafka_producer = KafkaProducer(**producer_kwargs)


[medium] resource-lifecycle

The module-level _kafka_producer singleton is never explicitly flushed or closed. With linger_ms=5, buffered messages can be lost on process shutdown. No Flask teardown hook or signal handler calls producer.flush() / producer.close().

Suggested fix: Register a Flask teardown_appcontext or atexit handler that calls _kafka_producer.flush() and _kafka_producer.close().

That's the tradeoff we have to accept for sending message async and not blocking the main thread for sending messages.

fullsend-ai-review · 2026-09-07T08:22:54Z

+    BatchRequestResponseList,
+)
+
+_kafka_producer: Optional[KafkaProducer] = None


[low] architectural-coherence

The Kafka path introduces a process-wide module-level singleton while the existing AMQP path creates a fresh connection per call. Tests work around this by directly resetting kafka_producer._kafka_producer = None. The singleton pattern is idiomatic for KafkaProducer but diverges from the project's existing connection model.

Suggested fix: Document the rationale for the singleton lifetime in a module-level docstring, or scope the producer to the Flask application context.

Creating Kafka connection for each message will be a heavy weight operation and considering the scale of request IIB receives, having a single Producer reduces the load at a good factor. So I will stick with this one.

fullsend-ai-review · 2026-09-07T08:22:54Z

+    global _kafka_producer
+
+    # First check without lock for performance
+    if _kafka_producer is not None:


[low] edge-case

Double-checked locking caches the producer indefinitely. No reconnection, cache-invalidation, or credential-rotation mechanism exists. The application must be fully restarted for new credentials or to recover from certain failure modes.

Suggested fix: Add a mechanism to invalidate the cached producer (e.g., TTL-based refresh, a reset function, or a signal handler).

It will become complex to add such facility in a single MR. It might be better to create a separate MR in case if the current implementation causes issue.

fullsend-ai-review · 2026-09-07T08:22:54Z

+                'security_protocol': 'SASL_SSL',
+                'sasl_mechanism': 'SCRAM-SHA-512',
+                'sasl_plain_username': username,
+                'sasl_plain_password': password,


[low] secrets-handling

Kafka password placed in producer_kwargs dict could be exposed if logger.exception() captures local variables in certain logging or APM configurations.

Suggested fix: Clear password from producer_kwargs in a finally block, or pass credentials as direct keyword arguments.

I don't think we are having any such custom logging configuration.

fullsend-ai-review · 2026-09-07T08:22:55Z


+def _build_request_state_change_data(
+    request: Request,
+) -> tuple:


[low] api-shape-patterns

_build_request_state_change_data and _build_batch_state_change_data return bare tuple instead of explicit Tuple[...] generics used elsewhere in the codebase.

Suggested fix: Use Tuple[BaseClassRequestResponse, Dict[str, Any]] and Optional[Tuple[...]].

fullsend-ai-review · 2026-09-07T08:22:55Z

    IIB_REQUEST_RELATED_BUNDLES_DIR: Optional[str] = None
    IIB_REQUEST_RECURSIVE_RELATED_BUNDLES_DIR: Optional[str] = None
    IIB_USER_TO_QUEUE: Union[Dict[str, str], Dict[str, Dict[str, str]]] = _get_empty_dict_str_str()
+    # Kafka messaging configuration


[low] naming-convention

Existing messaging config uses IIB_MESSAGING_* prefix; this PR introduces a parallel IIB_KAFKA_* namespace. A separate namespace is reasonable given the different transport backends, but diverges from the single-prefix pattern.

Suggested fix: Consider IIB_MESSAGING_KAFKA_* to keep messaging config under one prefix, or document the naming rationale.

As most of the configs are for Kafka, it suits better to have kafka for these configs.

fullsend-ai-review · 2026-09-07T08:22:55Z

        mock_sm.assert_not_called()


+@mock.patch('iib.web.messaging._send_kafka_messages')


[low] test-inadequate

test_send_messages_for_new_batch_of_requests_no_requests mocks _send_kafka_messages but does not assert it is NOT called when the request list is empty.

Suggested fix: Add mock_skm.assert_not_called().

fullsend-ai-review · 2026-09-07T08:22:55Z

-messages when a build request state changes and when a batch state changes. Please note that if a
-message can't be sent due to an infrastructure issue, the build request will continue as it is not
-considered a fatal error.
+IIB has support to send messages to an AMQP 1.0 broker (will be depricated in future) and to Apache Kafka topics.


[low] scope-creep

States AMQP 'will be deprecated' but the PR provides no deprecation warning, migration guide, or timeline.

Suggested fix: Remove the deprecation claim from the README or open a follow-up issue tracking AMQP deprecation.

No warning reqired, as the web service logs is only available to operator foundry team.

fullsend-ai-review · 2026-09-07T08:22:55Z

-messages when a build request state changes and when a batch state changes. Please note that if a
-message can't be sent due to an infrastructure issue, the build request will continue as it is not
-considered a fatal error.
+IIB has support to send messages to an AMQP 1.0 broker (will be depricated in future) and to Apache Kafka topics.


[low] documentation-quality

Contains 'depricated' (typo for 'deprecated') and 'Please notethat' (missing space between 'note' and 'that').

Suggested fix: Correct to 'deprecated' and 'Please note that'.

Thanks, will correct it.

ashwgit force-pushed the add-kafka-support branch from 6d3a2c4 to fb219fb Compare September 4, 2026 13:56

ashwgit force-pushed the add-kafka-support branch from fb219fb to 7dfa904 Compare September 4, 2026 14:16

ashwgit force-pushed the add-kafka-support branch from 7dfa904 to dce85fc Compare September 4, 2026 14:58

fullsend-ai-review Bot added the risk/moderate PR risk: moderate label Sep 4, 2026

ashwgit force-pushed the add-kafka-support branch from dce85fc to 45e62d0 Compare September 7, 2026 07:38

ashwgit force-pushed the add-kafka-support branch from 45e62d0 to 51173ac Compare September 7, 2026 07:58

fullsend-ai-review Bot added risk/elevated PR risk: elevated and removed risk/moderate PR risk: moderate labels Sep 7, 2026

fullsend-ai-review Bot added risk/moderate PR risk: moderate and removed risk/elevated PR risk: elevated labels Sep 7, 2026

fullsend-ai-review Bot reviewed Sep 7, 2026

View reviewed changes

fullsend-ai-review Bot suggested changes Sep 7, 2026

View reviewed changes

		mock_sm.assert_not_called()


		@mock.patch('iib.web.messaging._send_kafka_messages')

Uh oh!

Conversation

ashwgit commented Sep 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

fullsend-ai-review Bot commented Sep 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

fullsend-ai-review Bot commented Sep 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

fullsend-ai-review Bot commented Sep 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

fullsend-ai-review Bot commented Sep 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

fullsend-ai-review Bot commented Sep 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Review

Findings

Medium

Low

Review

Review

Uh oh!

fullsend-ai-review Bot commented Sep 4, 2026

Uh oh!

fullsend-ai-review Bot commented Sep 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

fullsend-ai-review Bot commented Sep 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

fullsend-ai-review Bot commented Sep 7, 2026

Uh oh!

fullsend-ai-review Bot commented Sep 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

fullsend-ai-review Bot left a comment

Choose a reason for hiding this comment

Uh oh!

fullsend-ai-review Bot left a comment

Choose a reason for hiding this comment

Uh oh!

fullsend-ai-review Bot Sep 7, 2026

Choose a reason for hiding this comment

Uh oh!

fullsend-ai-review Bot Sep 7, 2026

Choose a reason for hiding this comment

Uh oh!

ashwgit Sep 7, 2026

Choose a reason for hiding this comment

Uh oh!

fullsend-ai-review Bot Sep 7, 2026

Choose a reason for hiding this comment

Uh oh!

ashwgit Sep 7, 2026

Choose a reason for hiding this comment

Uh oh!

fullsend-ai-review Bot Sep 7, 2026

Choose a reason for hiding this comment

Uh oh!

ashwgit Sep 7, 2026

Choose a reason for hiding this comment

Uh oh!

fullsend-ai-review Bot Sep 7, 2026

Choose a reason for hiding this comment

Uh oh!

ashwgit Sep 7, 2026

Choose a reason for hiding this comment

Uh oh!

fullsend-ai-review Bot Sep 7, 2026

Choose a reason for hiding this comment

Uh oh!

fullsend-ai-review Bot Sep 7, 2026

Choose a reason for hiding this comment

Uh oh!

ashwgit Sep 7, 2026

Choose a reason for hiding this comment

ashwgit commented Sep 4, 2026 •

edited

Loading

fullsend-ai-review Bot commented Sep 4, 2026 •

edited

Loading

fullsend-ai-review Bot commented Sep 4, 2026 •

edited

Loading

fullsend-ai-review Bot commented Sep 4, 2026 •

edited

Loading

fullsend-ai-review Bot commented Sep 4, 2026 •

edited

Loading

fullsend-ai-review Bot commented Sep 4, 2026 •

edited

Loading

fullsend-ai-review Bot commented Sep 4, 2026 •

edited

Loading

fullsend-ai-review Bot commented Sep 7, 2026 •

edited

Loading

fullsend-ai-review Bot commented Sep 7, 2026 •

edited

Loading