Skip to content

feat: add Kafka publish for build/batch state-change events - #1381

Closed
ashwgit wants to merge 1 commit into
release-engineering:masterfrom
ashwinifork:add-kafka-support
Closed

ashwgit wants to merge 1 commit into
release-engineering:masterfrom
ashwinifork:add-kafka-support

Conversation

@ashwgit

@ashwgit ashwgit commented Sep 4, 2026 •

Copy link
Copy Markdown
Contributor

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

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

Copy link
Copy Markdown

🤖 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

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

Copy link
Copy Markdown

🤖 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

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

Copy link
Copy Markdown

🤖 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 fullsend-ai-review Bot added the risk/moderate PR risk: moderate label Sep 4, 2026
@fullsend-ai-review

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

Copy link
Copy Markdown

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

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

Copy link
Copy Markdown

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

Copy link
Copy Markdown

/fs-review

@fullsend-ai-review

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

Copy link
Copy Markdown

🤖 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

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

Copy link
Copy Markdown

🤖 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 fullsend-ai-review Bot added risk/elevated PR risk: elevated and removed risk/moderate PR risk: moderate labels Sep 7, 2026
@fullsend-ai-review

Copy link
Copy Markdown

/fs-review

@fullsend-ai-review

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

Copy link
Copy Markdown

🤖 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 fullsend-ai-review Bot added risk/moderate PR risk: moderate and removed risk/elevated PR risk: elevated labels Sep 7, 2026

@fullsend-ai-review fullsend-ai-review Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 fullsend-ai-review Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See the review comment for full details.

Comment thread iib/web/messaging.py
if batch_data:
content, properties = batch_data
send_kafka_message(producer, batch_topic, content, properties)
except: # noqa: E722

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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:.

Comment thread iib/web/kafka_producer.py
if ca_file:
producer_kwargs['ssl_cafile'] = ca_file

_kafka_producer = KafkaProducer(**producer_kwargs)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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().

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Comment thread iib/web/kafka_producer.py
BatchRequestResponseList,
)

_kafka_producer: Optional[KafkaProducer] = None

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment thread iib/web/kafka_producer.py
global _kafka_producer

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

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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).

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment thread iib/web/kafka_producer.py
'security_protocol': 'SASL_SSL',
'sasl_mechanism': 'SCRAM-SHA-512',
'sasl_plain_username': username,
'sasl_plain_password': password,

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Comment thread iib/web/messaging.py

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

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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[...]].

Comment thread iib/web/config.py
mock_sm.assert_not_called()


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

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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().

Comment thread README.md
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.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Comment thread README.md
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.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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'.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, will correct it.

@ashwgit

ashwgit commented Sep 8, 2026

Copy link
Copy Markdown
Contributor Author

As per discussion I will create this MR against main branch and close this MR.

@ashwgit ashwgit closed this Sep 8, 2026
@fullsend-ai-retro

fullsend-ai-retro Bot commented Sep 8, 2026 •

Copy link
Copy Markdown

🤖 Finished Retro · ✅ Success · Started 3:54 PM UTC · Completed 4:06 PM UTC

Commit: 51173ac · View workflow run →

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

@fullsend-ai-retro

Copy link
Copy Markdown

Retro: PR #1381 — Kafka Publish Support

Timeline

PR #1381 (human-authored by @ashwgit) added Kafka messaging support alongside existing AMQP in IIB. The review agent ran 6 times over 4 days before producing a usable review:

Run Commit Result Cost
33873033239 6d3a2c4 Timeout (2×20min) $4.61
33880821239 fb219fb Timeout (2×20min) —
33882821296 7dfa904 Stale-head discard $6.82
33886792106 dce85fc Timeout (2×20min) —
34096472468 45e62d0 Stale-head discard $6.43
34098135662 51173ac Final review posted $6.23

Known cost: $24.09 (only $6.23 produced the usable review). 12 additional no-op pull_request_review runs were triggered by comment activity. The PR was closed without merging — author decided to retarget to a different branch.

Review Finding Quality

The final review identified 15 findings (2 medium, 13 low). Author accepted 2 of 15 (13%):

  • Accepted (2): typo fix, f-string→%-style logging format
  • Rejected (7): design tradeoffs the author had already considered (singleton lifecycle, async flush, function-level imports, naming prefix, secrets exposure, TLS fallback, deprecation scope)
  • Deferred (1): credential rotation (acknowledged, punted to follow-up)
  • No response (5): bare except:, docstring params, type annotations, return types, test assertion

The 2 medium findings (bare except: catching BaseException, producer never flushed/closed) were the most valuable, but got no uptake — one unanswered, one rejected with justification about async tradeoffs. The 13 low findings were primarily convention/style preferences where the author had informed reasons to disagree.

Evidence for Existing Issues

All identified improvement opportunities are covered by existing tracked issues. This PR adds supporting evidence:

Platform (fullsend-ai/fullsend):

  • Stale-head waste (#1331, #2399): 2 successful reviews discarded as stale-head, costing $13.25 in wasted compute.
  • Timeout on complex PRs (#7042, #7023): 3 of 6 runs timed out at the 20-min iteration limit. The agent was actively analyzing code (dispatched 5 sub-agents, mid-analysis) when killed.
  • Redundant runs (#4681, #5139): 6 review runs to produce 1 usable review.
  • Bot event filtering (#4984, #2994): 12 no-op workflow runs triggered by author reply comments.
  • Scope heuristic (#6891): A pre-flight complexity check could have adjusted the budget for this multi-file Kafka integration, potentially avoiding timeouts.

Agents (fullsend-ai/agents):

  • Low-severity finding volume (#1105): 13 low-severity findings with near-zero acceptance rate. Filtering these would have improved signal-to-noise.
  • Verdict calibration (#1141): CHANGES_REQUESTED verdict was driven primarily by low-severity convention findings that the author rejected; COMMENT or APPROVE-with-suggestions may have been more appropriate.
  • Verbose low-value output (#370): 13% acceptance rate indicates the review was generating more noise than signal for this codebase.
  • Self-contradicting findings (#1106): Some findings (e.g., TLS config, import organization) acknowledged the code's rationale in the finding text but still flagged them.

Source repo (release-engineering/iib):

  • Domain-specific guidance (#1366): The low acceptance rate was largely driven by convention/style findings that conflicted with project-specific design decisions (singleton Kafka producer, function-level imports for background threads, IIB_KAFKA_* naming prefix). An AGENTS.md documenting these conventions would prevent these findings.

Assessment

No new proposals are warranted — the improvement opportunities surfaced by this PR are well-tracked across the platform, agents, and source repos. The highest-impact fixes would be resolving stale-head waste (#2399) and filtering low-severity findings (#1105), which together would have reduced this PR's review cost from $24+ to ~$6 and the finding volume from 15 to 2–3 high-confidence items.

@ashwgit

ashwgit commented Sep 9, 2026

Copy link
Copy Markdown
Contributor Author

PR against main branh - #1382

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

risk/moderate PR risk: moderate

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant