EBX Kafka Pub/Sub with the ps-mdm Framework

Download the working client

The companion client is available as a thin adapter JAR or as a bundled JAR containing the Apache Kafka runtime dependencies. Both downloads, the example properties file, SHA-256 checksums, source code, compatibility boundary, and deployment instructions are on the product page. The public source repository is KinshukON/ps-mdm-kafka-client.

What ps-mdm already gives you

The framework separates EBX-facing behavior from broker-specific behavior. MessagingConfiguration loads configuration and creates a MessagingClient. That facade transforms EBX Adaptation records into JSON, applies retry logic, and delegates transport calls to a vendor adapter implementing MessagingClientInterface. Kafka, RabbitMQ, or another transport can therefore sit behind the same EBX service.

The outbound message contract

The default outbound transformer wraps a record in two objects. messageMetadata identifies the EBX dataspace, dataset, and table; messagePayload contains the JSON representation produced by JSONUtil.jsonObjectForAdaptation(adaptation, true). A representative event is:

{
  "messageMetadata": {
    "messageDataspace": "Reference",
    "messageDataset": "CustomerMaster",
    "messageTable": "/root/Customer"
  },
  "messagePayload": {
    "customerId": "C-1042",
    "legalName": "Northstar Components",
    "status": "ACTIVE"
  }
}

This is a useful transport envelope, but it is not yet a durable enterprise event contract. Add an event ID, event type, schema version, occurred-at time, source commit or changeset identifier, operation (CREATE, UPDATE, DELETE), record primary key, actor/service identity, and trace ID. Keep broker metadata such as partition and offset outside the business payload.

Configure the framework—and mind the exact keys

The framework can read the main EBX properties file or an alternate file named by the JVM system property ebx.messaging.properties. Java property names are case-sensitive. The reviewed code looks up ebx.messaging.configurationclassname in lowercase, even though the sample properties document shows ebx.messaging.configurationClassName. Use the runtime constant’s lowercase spelling unless you deliberately correct and test the framework.

ebx.messaging.enable=true
ebx.messaging.configurationclassname=com.acme.ebx.kafka.KafkaMessagingConfiguration
ebx.messaging.clientclassname=com.acme.ebx.kafka.KafkaClient

# Existing ps-mdm properties
ebx.messaging.host=kafka-1.internal.example
ebx.messaging.port=9093
ebx.messaging.inbound.destinationname=mdm.customer.command.v1
ebx.messaging.outbound.destinationname=mdm.customer.event.v1
ebx.messaging.connection.timeout=10000
ebx.messaging.connection.retry=3
ebx.messaging.connection.retry.delay=1000
ebx.messaging.connection.reuse=true
ebx.messaging.outbound.useprettyprint=false

# Extension properties read by KafkaMessagingConfiguration
ebx.messaging.kafka.bootstrap.servers=kafka-1.internal.example:9093,kafka-2.internal.example:9093
ebx.messaging.kafka.group.id=ebx-customer-command-v1
ebx.messaging.kafka.client.id=ebx-mdm
ebx.messaging.kafka.security.protocol=SASL_SSL
ebx.messaging.kafka.sasl.mechanism=SCRAM-SHA-512
ebx.messaging.kafka.acks=all
ebx.messaging.kafka.enable.idempotence=true

Do not keep production passwords directly in this file. The existing loader treats only ebx.messaging.password as secret for logging purposes; that is not a secret store. Resolve credentials through the application server or an approved secrets provider and keep JAAS strings, keystore passwords, and OAuth client secrets out of source control and diagnostic output.

Implement the missing Kafka adapter

The adapter must expose the exact constructor used by reflection: one argument of type MessagingConfiguration. Keep a thread-safe KafkaProducer<String,String> for outbound traffic. A KafkaConsumer is not thread-safe, so create and use it inside one controlled subscriber worker or guard it with strict single-thread ownership.

public final class KafkaClient extends AbstractMessagingClientInterface {
  private final KafkaProducer<String, String> producer;
  private final KafkaMessagingConfiguration config;

  public KafkaClient(MessagingConfiguration base) {
    super(base);
    this.config = (KafkaMessagingConfiguration) base;
    this.producer = new KafkaProducer<>(config.producerProperties());
  }

  @Override
  public void publishMessage(String topic, OutboundMessage message) throws Exception {
    String key = message.getHeaderProperty("recordKey");
    ProducerRecord<String, String> record =
        new ProducerRecord<>(topic, key, message.getBody());
    record.headers().add("correlationId", bytes(message.getCorrelationID()));
    producer.send(record).get(config.deliveryTimeoutMs(), TimeUnit.MILLISECONDS);
  }

  @Override
  public List<InboundMessage> subscribe(String topic, int timeoutMs) {
    try (KafkaConsumer<String, String> consumer =
             new KafkaConsumer<>(config.consumerProperties())) {
      consumer.subscribe(List.of(topic));
      ConsumerRecords<String, String> records =
          consumer.poll(Duration.ofMillis(timeoutMs));
      List<InboundMessage> result = new ArrayList<>();
      for (ConsumerRecord<String, String> record : records) {
        InboundMessage message = new InboundMessage();
        message.setBody(record.value());
        message.setMessageId(topic + "-" + record.partition() + "-" + record.offset());
        result.add(message);
      }
      return result;
    }
  }

  @Override public void shutdown() { producer.flush(); producer.close(); }
}

The sketch intentionally blocks on the producer acknowledgement because the ps-mdm facade treats a returned call as success. In a higher-throughput service, collect asynchronous acknowledgements and expose failure deterministically; never log “sent” before Kafka has accepted the record under the selected acknowledgement policy. Implement the remaining interface methods with a documented behavior. For Kafka, queue-oriented methods can throw UnsupportedOperationException; topic listing can use AdminClient; createConnection and resetConnection should reflect Kafka client lifecycle rather than imitate a JMS connection.

Publish from EBX

PublishSelectedRecordsService already demonstrates the happy path: execute the selected-record request, create an OutboundMessage, set its delivery time and Adaptation, then call messageClient.publishMessage(null, message). Passing null chooses the configured outbound destination. The default transformer serializes the adaptation immediately before delegation.

MessagingClient client = MessagingConfiguration.getMessagingClient();
OutboundMessage event = new OutboundMessage();
event.setAdaptation(customerRecord);
event.setDeliveryTime(System.currentTimeMillis());
event.setCorrelationID(traceId);
event.addHeaderProperty("recordKey", customerId);
client.publishMessage("mdm.customer.event.v1", event);

Use the stable master-record key as the Kafka key when per-entity ordering matters. Do not use a random key for updates to the same customer. The stock user service is explicitly example code: it catches all exceptions and prints a stack trace, which is insufficient for a governed production operation. Surface an EBX OperationException or durable publication status, authorize the service, audit the action, and avoid exposing restricted fields in the serialized adaptation.

Subscribe and apply changes in EBX

MessagingClient.subscribe(topic, timeout) delegates to the adapter and runs each body through MessageInboundTransformationImpl, returning a list of maps. That is only deserialization. It does not locate a dataset, validate a table, create a procedure, or commit a record. Your subscriber application layer must perform those EBX steps explicitly.

1. Run a single controlled consumer worker per consumer instance—typically an application-server-managed scheduled task or service, not an unbounded thread created by a table trigger.
2. Poll a bounded batch and validate envelope type, schema version, source, topic, and allowed EBX target.
3. Resolve the target dataspace, dataset, and table from an allowlist. Never trust arbitrary paths from the message.
4. Execute the write in an EBX Procedure with an authenticated service session. Apply table constraints, permissions, and business validation.
5. Store the Kafka event ID or topic-partition-offset in an EBX integration ledger in the same EBX transaction as the business write.
6. Commit the Kafka offset only after the EBX transaction succeeds. On a transient failure, leave the offset uncommitted. On a permanent validation failure, publish a sanitized failure event or dead-letter record, then commit according to the documented policy.

This produces at-least-once delivery with idempotent application. Kafka’s idempotent producer prevents certain producer retry duplicates; it does not make a Kafka consume-and-EBX-write operation atomic. The integration ledger closes that gap by making duplicate delivery detectable inside EBX.

Transaction boundary: prefer an outbox for automatic publication

A synchronous call from an EBX table trigger to Kafka creates a dual-write problem: the EBX transaction may commit after Kafka fails, or Kafka may accept the event before EBX rolls back. For automatic publication, write an outbox row inside the same EBX procedure as the master-data change. A scheduled publisher reads pending outbox rows, publishes with a deterministic event ID, records the acknowledgement, and marks the row delivered. Consumers still remain idempotent.

Use the existing manual publish service for deliberate operator-driven exports or controlled testing. Use an outbox when the business contract says every committed change must eventually produce an event.

Source-level issues to address before production

• Inbound fallback: the reviewed recieveMessages method falls back to getOutboundDestination() when the inbound destination argument is empty. Correct it to getInboundDestination() and cover the behavior with a regression test.
• Body-only publication: PublishUtils.publishDataspaceDiff sets a string body but calls the default publish path, which attempts to transform a null Adaptation. Add a pass-through transformer or a body-only overload before using this utility.
• Retry meaning: ebx.messaging.connection.retry controls total loop attempts in the reviewed code, despite being described as retries. Set and document it accordingly or correct the loop.
• Failure reporting: the sample publish service swallows failures after printStackTrace(). Replace that with structured logging, an EBX-visible failure, and operational metrics.
• Configuration drift: sample documentation refers to older class names and a mixed-case property key. Treat compiled constants and tested runtime configuration as authoritative.
• Payload governance: the generic adaptation serializer may include fields that should not leave EBX. Use a purpose-built outbound transformation with an explicit allowlist and schema.

Production checklist

• Package ps-mdm, the Kafka adapter JAR, and version-compatible Kafka client dependencies on the application-server classpath without duplicate client versions.
• Confirm the EBX module registers the intended user service or scheduled subscriber and invokes shutdown on undeploy.
• Create topics with reviewed partition, replication, retention, compaction, and authorization settings.
• Use TLS and workload identities; test credential and certificate rotation.
• Validate schemas and compatibility before deploy; quarantine malformed or unsupported versions.
• Measure publish acknowledgements, error and retry rates, consumer lag, EBX apply latency, rejected records, dead-letter volume, and end-to-end freshness.
• Test broker loss, rebalance, duplicate delivery, poison records, EBX rollback, restart after publish-before-status-update, and replay from the oldest retained event.
• Reconcile Kafka events against the EBX integration ledger and business records.

What this design achieves

The ps-mdm framework is a useful anti-corruption layer: EBX services work with EBX records and framework messages, while a replaceable adapter owns Kafka APIs. With a Kafka adapter, explicit event contract, keyed ordering, an outbound outbox, an inbound idempotency ledger, secured configuration, and failure tests, it becomes a credible publish/subscribe bridge rather than a demo that merely sends JSON.

Use OffsetDock to inspect the topics, keys, payload envelopes, partitions, and consumer lag during development. Validate EBX APIs and deployment behavior against the exact supported release in the official TIBCO EBX documentation, and Kafka properties against the official Apache Kafka documentation.