Kafka Intro

Apache Kafka® (https://kafka.apache.org) is a distributed streaming platform with three key capabilities:   

Kafka is generally used for two broad classes of applications:   

Kafka concepts:   

Kafka has four core APIs:     

FAJ Kafka adapter

FAJ Kafka adapter is designed to send and receive FIX messages to or from Kafka and connect FIX sources with Kafka. It uses Kafka Producer and Consumer API to communicate with Kafka cluster. 

FAJ Kafka adapter provides a few main interfaces.

Producer interface is responsible for sending FIX messages to Kafka topics and Consumer - for receiving. Both of them can be instantiated via a factory, which is provided with the adapter:

// prepare Kafka client factory
Config config = new Config("kafka", "kafka-adapter.propeties");
ClientFactory clientFactory = ClientFactory.getInstance();

// Create producer instance with id `KProducer`
Producer producer = clientFactory.createProducerClient("KProducer", outConfig);
// send FIX message to Kafka

// Create producer instance with id `KConsumer`
Consumer consumer = clientFactory.createConsumerClient("KConsumer", inConfig);
// register a callback for received messages
consumer.setMessageListener((sessionID, message) ->
                    System.out.println("Message received by client '" + sessionID + "' - " + new String(message)));

The minimal kafka-adapter.propeties file should include a list of Kafka adapter clients and link them to corresponding Kafka topics:

# list of kafka clients (consumers and producer)
kafka.clients = KProducer, KConsumer

# main properties to specify producer's client id and topic
kafka.producer.KProducer.client.id = KProducer
kafka.producer.KProducer.topic = PTopic1

# main properties to specify consumer's client id and topic
kafka.consumer.KConsumer.client.id = KConsumer
kafka.consumer.KConsumer.topics = KTopic2

Each Kafka adapter client (producer or consumer) can be additionally configured with original Kafka Producer or Consumer configuration options if they are added after the client' prefix:

# Apply batch size for KProducer producer client only
kafka.producer.KProducer.batch.size = 16384

# Apply memory bufer size for every producer
kafka.producer.buffer.memory = 33554432

Pair of producers and consumers may be linked with FixClientAdaptor interface for easier communication management. In this case, a pair of producer and consumer may be initialized and closed at the same time:

Producer producer = clientFactory.createProducerClient("KProducer", outConfig);
Consumer consumer = clientFactory.createConsumerClient("KConsumer", inConfig);
FixClientAdaptor adaptor = new FixClientAdaptorImpl(consumer, producer);
adaptor.setMessageListener((sessionID, message) -> System.out.println("Message received by client '" + sessionID + "' - " + message.toPrintableString()));

FAJ Kafka Adapter Samples 

FAJ package includes samples to demonstrate communication between FIX engine and Kafka. The quick start demo consists of a few components:

Steps to run FAJ Kafka adapter demo

  1. Install and start Kafka:
  2. Download and unpack the latest FAJ Kafka distribution package (fixaj-kafka-distribution-<version>-bin.zip).

    It will contain the following folders/parts:
  3. Start FIX server sample to accept FIX connection from FixKafkaBridge (server sends 10 test FIX messages to KafkaBridge after establishing FIX connection):
    > examples/bin/runFixAcceptorServer.bat (Windows)
    > examples/bin/runFixAcceptorServer (Linux)
  4.  Start KafkaClientRouter to route messages from 'producerTopic' to 'consumerTopic' to provide messages back to KafkaBridge:
    > examples/bin/runKafkaClientRouter.bat (Windows)
    > examples/bin/runKafkaClientRouter (Linux)
  5. Start FixKafkaBridge with properties from "tools/etc" folder:
    > tools/bin/startKafkaBridge.bat (Windows)
    > tools/bin/startKafkaBridge (Linux)

As a result of these steps, you will be able to see that: