Hazelcast

A Kafka Connect sink connector for writing records from Kafka to Hazelcast.

KCQL support 

The following KCQL is supported:

INSERT INTO <hazelcast-namespace> 
SELECT FIELD, ... 
FROM <your-kafka-topic> 
[PK FIELD, ...] 
WITHFORMAT JSON|AVRO 
STOREAS RELIABLE_TOPIC|RING_BUFFER|QUEUE|SET|LIST|IMAP|MULTI_MAP|ICACHE

Example:

-- store into a ring buffer
INSERT INTO lenses SELECT x,y,z FROM topicB WITHFORMAT AVRO STOREAS RING_BUFFER
-- store into a reliable topic
INSERT INTO lenses SELECT x,y,z FROM topicB WITHFORMAT AVRO STOREAS RELIABLE_TOPIC
-- store into a queue
INSERT INTO lenses SELECT x,y,z FROM topicB WITHFORMAT AVRO STOREAS QUEUE
-- store into a set
INSERT INTO lenses SELECT x,y,z FROM topicB WITHFORMAT AVRO STOREAS SET
-- store into a list
INSERT INTO lenses SELECT x,y,z FROM topicB WITHFORMAT AVRO STOREAS LIST
-- store into an i-map with field1 used as the map key
INSERT INTO lenses SELECT x,y,z FROM topicB PK field1 WITHFORMAT AVRO STOREAS IMAP
-- store into a multi-map with field1 used as the map key
INSERT INTO lenses SELECT x,y,z FROM topicB PK field1 WITHFORMAT AVRO STOREAS MULTI_MAP
-- store into an i-cache with field1 used as the cache key
INSERT INTO lenses SELECT x,y,z FROM topicB PK field1 WITHFORMAT AVRO STOREAS ICACHE

Concepts 

The connector takes the value from the Kafka Connect SinkRecords and inserts/updates an entry in HazelCast. The Sink supports writing to

• reliable topics
• ring buffers
• queues
• sets
• lists
• imap
• multi-map
• icache.

Primary Keys 

When inserting into MAP or MULTI_MAP we need a key, the PK keyword can be used to specify the fields which will be used for the key value. The field values will be concatenated and separated by a -. If no fields are set the topic name, partition and message offset are used.

With Format 

Hazelcast requires that data stored in collections and topics is serializable. The Sink offers two modes to store data.

AVRO In this mode the Sink converts the SinkRecords from Kafka to AVRO encoded byte arrays. JSON In this mode the Sink converts the SinkRecords from Kafka to JSON strings.

This behavior is controlled by the KCQL statement in the connect.hazelcast.kcql option. The default is JSON.

Stored As 

The Hazelcast Sink supports storing data in RingBuffers, ReliableTopics, Queues, Sets, Lists, IMaps, Multi-maps and ICaches. This behaviour is controlled by the KCQL statement in the connect.hazelcast.kcql option.

Parallel Writes 

By default each task in the Sink will write the records it receives sequentially, the Sink optionally supports parallel writes where an executorThreadPool is started and records are written in parallel. While this results in better performance we can’t guarantee the order of the writes.

To enable parallel writes set the connect.hazelcast.parallel.write configuration option to true

Error polices 

The connector supports Error polices .

Quickstart 

Launch the stack 

1. Copy the docker-compose file.
2. Bring up the stack.

export CONNECTOR=hazelcast
docker-compose up -d hazelcast

Next configure and start Hazelcast so that the Connector can join the cluster. Login to the Hazelcast container and create a file called hazelcast.xml with the following content:

docker exec -ti hazelcast /bin/bash 
apt-get update && apt-get install -y vim

<hazelcast xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
           xsi:schemaLocation="http://www.hazelcast.com/schema/config/hazelcast-config-3.8.xsd"
           xmlns="http://www.hazelcast.com/schema/config">
    <cluster-name>dev</cluster-name>
</hazelcast>

Then start Hazelcast:

hz start -c hazelcast.xml

Start the connector 

If you are using Lenses, login into Lenses and navigate to the connectors page , select Hazelcast as the sink and paste the following:

name=hazelcast
connector.class=com.datamountaineer.streamreactor.connect.hazelcast.sink.HazelCastSinkConnector
tasks.max=1
topics=orders
connect.hazelcast.cluster.members=hazelcast
connect.hazelcast.cluster.name=dev
connect.hazelcast.kcql=INSERT INTO orders SELECT * FROM orders WITHFORMAT JSON STOREAS QUEUE

To start the connector without using Lenses, log into the fastdatadev container:

docker exec -ti fastdata /bin/bash

and create a connector.properties file containing the properties above.

Create the connector, with the connect-cli :

connect-cli create hazelcast < connector.properties

Wait a for the connector to start and check its running:

connect-cli status hazelcast

Inserting test data 

In the to fastdata container start the kafka producer shell:

kafka-avro-console-producer \
 --broker-list localhost:9092 --topic orders \
 --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"id","type":"int"},{"name":"created","type":"string"},{"name":"product","type":"string"},{"name":"price","type":"double"}, {"name":"qty", "type":"int"}]}'

the console is now waiting for your input, enter the following:

{"id": 1, "created": "2016-05-06 13:53:00", "product": "OP-DAX-P-20150201-95.7", "price": 94.2, "qty":100}

Check for data in Hazelcast 

Login to the hazelcast container:

 docker exec -ti hazelcast /bin/bash

Run start the console app:

 java -cp hazelcast-all-3.8.4.jar com.hazelcast.client.console.ClientConsoleApp

In the app, switch to the dev namespace and check the queue:

ns dev

q.iterator

Clean up 

Bring down the stack:

docker-compose down

Options