Class KafkaAuditSink

java.lang.Object
com.mcpdbwizard.pub.KafkaAuditSink
All Implemented Interfaces:
McpAuditSink
public class KafkaAuditSink extends Object implements McpAuditSink
Reference McpAuditSink that publishes audit records to a Kafka topic.

Select it with MCP_AUDIT_SINK=com.mcpdbwizard.pub.KafkaAuditSink.

Environment
MCP_AUDIT_KAFKA_BOOTSTRAPRequired. Bootstrap servers.
MCP_AUDIT_KAFKA_TOPICTopic, default "mcp-audit".
MCP_AUDIT_KAFKA_ON_FULLblock (default) or drop.
MCP_AUDIT_KAFKA_BLOCK_MSHow long block waits, default "5000".

What this does and does not guarantee — read before relying on it

This is a reference implementation, not a durable audit pipeline. Buffering is the Kafka producer's in-memory buffer. If the broker is unreachable for longer than that buffer and the configured block time absorb, records are dropped — counted and reported, but gone. Nothing here spools to disk and replays, and a disk spool is the substantial work that a genuinely durable trail needs (see docs/mcp-audit-sink-plan.md §2.3).

The two policies express the honest trade:

  • block — a tool call waits up to MCP_AUDIT_KAFKA_BLOCK_MS for buffer space, so a broker outage slows the database work. Closer to audit semantics, at the cost of coupling Oracle calls to Kafka's availability.
  • drop — the call never waits, and the record is lost on a full buffer. Correct when the trail is valuable but not load-bearing; wrong when somebody will later rely on it being complete.

acks=all is set, so a record acknowledged by the broker is on every in-sync replica. That covers durability after hand-off; it says nothing about records that never got there. Copyright 2003-2026 ATB Consultancy Services Ltd (formerly Orinda Software Ltd, Dublin, Ireland)

  • Field Details

  • Constructor Details

  • Method Details

    • topicFrom

      public static String topicFrom(String theSetting)
      The configured topic, or "mcp-audit" when the setting is absent or blank.

      Public so a status page can show the topic records will actually go to, rather than restating this default somewhere else and drifting from it.

    • record

      public void record(McpAuditEvent theEvent)
      Record one call. Must not throw.

      Never throws: this runs in a finally on the tool-call path, and an audit failure must not replace the caller's real result.

      Specified by:
      record in interface McpAuditSink
      Parameters:
      theEvent - the call to record; never null

    • flush

      public boolean flush()
      Confirm that everything handed to McpAuditSink.record(com.mcpdbwizard.pub.McpAuditEvent) since the last flush is durably accepted.

      This is what makes a spool possible. A spool writes each record to disk first and may only delete it once the sink has genuinely taken it — and for an asynchronous sink like Kafka, record returning tells you nothing, because the send has not completed yet.

      The default returns true, which is correct for a sink that delivers synchronously inside record. An asynchronous sink must override it, and must return false if anything since the last flush was lost — returning true on a failed batch would let a spool delete records that never arrived, which is the one way a spool can be worse than no spool.

      Blocks until the producer has completed every outstanding send, then reports whether any of them failed. A spool depends on this answer being honest: saying "delivered" about a batch that did not arrive would let it delete records that were never received.

      Specified by:
      flush in interface McpAuditSink
      Returns:
      true if everything since the last flush is safely delivered

    • getDroppedCount

      public long getDroppedCount()
      How many records have been lost. Non-zero means the trail has holes.
      Specified by:
      getDroppedCount in interface McpAuditSink

    • close

      public void close()
      Specified by:
      close in interface McpAuditSink