Kafka Connect CoAP Source

A Connector and Source to stream messages from a CoAP server and write them to a Kafka topic.

The Source supports:

  1. DTLS secure clients.
  2. Observable resources.

The Source Connector automatically converts the CoAP response into a Kafka Connect Struct to be store in Kafka as Avro or Json dependent on the Converters used in Connect. The schema can found here.

The key of the Struct message sent to Kafka is made from the source defined in the message, the resource on the CoAP server and the message id.


  • Confluent 3.2
  • Java 1.8
  • Scala 2.11


Confluent Setup

Follow the instructions here.

CoAP Setup

The connector uses Californium Java API under the hood. A resource is host at coap://californium.eclipse.org:5683. Copper, a FireFox browser addon is available so you can browse the server and resources.. This is an observable non confirmable resource. You can view messages via the browser addon by selecting the resource and clicking the observe button on the top menu.

Source Connector QuickStart

We will start the connector in distributed mode. Each connector exposes a rest endpoint for stopping, starting and updating the configuration. We have developed a Command Line Interface to make interacting with the Connect Rest API easier. The CLI can be found in the Stream Reactor download under the bin folder. Alternatively the Jar can be pulled from our GitHub releases page.

Starting the Connector

Download, unpack and install the Stream Reactor. Follow the instructions here if you haven’t already done so. All paths in the quickstart are based in the location you installed the Stream Reactor.

Start Kafka Connect in distributed more by running the start-connect.sh script in the bin folder.

➜ bin/start-connect.sh

Once the connector has started we can now use the kafka-connect-tools cli to post in our distributed properties file for MQTT. If you are using the dockers you will have to set the following environment variable to for the CLI to connect to the Rest API of Kafka Connect of your container.

export KAFKA_CONNECT_REST="http://myserver:myport"
➜  bin/cli.sh create coap-source < conf/coap-source.properties

#Connector name=`coap-source`
name = coap-source
tasks = 1
connector.class = com.datamountaineer.streamreactor.connect.coap.source.CoapSourceConnector
connect.coap.uri = coap://californium.eclipse.org:5683
connect.coap.kcql = INSERT INTO coap-topic SELECT * FROM obs-pumping-non
#task ids: 0

The coap-source.properties file defines:

  1. The name of the source.
  2. The name number of tasks.
  3. The class containing the connector.
  4. The uri of the CoAP Server and port to connect to.
  5. The KCQL routing querying.. This specifies the target topic and source resource on the CoAP server.

If you switch back to the terminal you started Kafka Connect in you should see the CoAP Source being accepted and the task starting.

We can use the CLI to check if the connector is up but you should be able to see this in logs as-well.

#check for running connectors with the CLI
➜ bin/cli.sh ps
    ____        __        __  ___                  __        _
   / __ \____ _/ /_____ _/  |/  /___  __  ______  / /_____ _(_)___  ___  ___  _____
  / / / / __ `/ __/ __ `/ /|_/ / __ \/ / / / __ \/ __/ __ `/ / __ \/ _ \/ _ \/ ___/
 / /_/ / /_/ / /_/ /_/ / /  / / /_/ / /_/ / / / / /_/ /_/ / / / / /  __/  __/ /
/_____/\__,_/\__/\__,_/_/  /_/\____/\__,_/_/ /_/\__/\__,_/_/_/ /_/\___/\___/_/
       ______                 _____
      / ____/___  ____ _____ / ___/____  __  _______________
     / /   / __ \/ __ `/ __ \\__ \/ __ \/ / / / ___/ ___/ _ \ By Andrew Stevenson
    / /___/ /_/ / /_/ / /_/ /__/ / /_/ / /_/ / /  / /__/  __/
    \____/\____/\__,_/ .___/____/\____/\__,_/_/   \___/\___/
                    /_/ (com.datamountaineer.streamreactor.connect.coap.source.CoapSourceTask:54)
[2017-01-09 20:42:44,830] INFO CoapConfig values:
    connect.coap.uri = coap://californium.eclipse.org:5683
    connect.coap.bind.port = 0
    connect.coap.truststore.pass = [hidden]
    connect.coap.cert.chain.key = client
    connect.coap.keystore.path =
    connect.coap.kcql = INSERT INTO coap-topic SELECT * FROM obs-pumping-non
    connect.coap.truststore.path =
    connect.coap.certs = []
    connect.coap.keystore.pass = [hidden]
    connect.coap.bind.host = localhost
[2017-01-09 20:42:44,831] INFO Source task WorkerSourceTask{id=coap-source-0} finished initialization and start (org.apache.kafka.connect.runtime.WorkerSourceTask:138)
[2017-01-09 20:42:45,927] INFO Discovered resources /.well-known/core (com.datamountaineer.streamreactor.connect.coap.source.CoapReader:60)
[2017-01-09 20:42:45,927] INFO Discovered resources /large (com.datamountaineer.streamreactor.connect.coap.source.CoapReader:60)
[2017-01-09 20:42:45,928] INFO Discovered resources /large-create (com.datamountaineer.streamreactor.connect.coap.source.CoapReader:60)
[2017-01-09 20:42:45,928] INFO Discovered resources /large-post (com.datamountaineer.streamreactor.connect.coap.source.CoapReader:60)
[2017-01-09 20:42:45,928] INFO Discovered resources /large-separate (com.datamountaineer.streamreactor.connect.coap.source.CoapReader:60)
[2017-01-09 20:42:45,928] INFO Discovered resources /large-update (com.datamountaineer.streamreactor.connect.coap.source.CoapReader:60)

Check for records in Kafka

Check for records in Kafka with the console consumer.

➜  bin/kafka-avro-console-consumer \
   --zookeeper localhost:2181 \
   --topic coap-topic \


  1. Secure DTLS client connection.
  2. Supports Observable resources to stream changes on a resource to Kafka.
  3. Routing of data via KCQL to topics.
  4. Automatic conversion of CoAP Response messages to Connect Structs.

Kafka Connect Query Language

K afka C onnect Q uery L anguage found here GitHub repo allows for routing and mapping using a SQL like syntax, consolidating typically features in to one configuration option.

The CoAP Source supports the following:

INSERT INTO <topic> SELECT * FROM <resource>

No selection of fields on the CoAP message is support. All the message attributes are mapped to predefined Struct representing the CoAP response message.

DTLS Client

The Connector use the Californium Java API and for secure connections use the Scandium security module provided by Californium. Scandium (Sc) is an implementation of Datagram Transport Layer Security 1.2, also known as RFC 6347.

Please refer to the Californium certification repo page for more information.

DTLS Client connections can be enabled by setting the connect.coap.keystore.pass property. If set you must provide the following or the Connector will not start and throw a configuration exception:

  • connect.coap.keystore.pass
  • connect.coap.keystore.path
  • connect.coap.truststore.pass
  • connect.coap.truststore.path


The key and truststore must be available on the local disk of the worker task.

Loading specific certificates can be achieved by providing a comma separated list for the connect.coap.certs configuration option. The certificate chain can be set by the connect.coap.cert.chain.key configuration option.



Uri of the CoAP server.

  • Data Type : string
  • Importance: high
  • Optional : no


The KCQL statement to select and route resources to topics.

  • Data Type : string
  • Importance: high
  • Optional : no


The port the DTLS connector will bind to on the Connector host.

  • Data Type : int
  • Importance: medium
  • Optional : yes
  • Default : 0


The hostname the DTLS connector will bind to on the Connector host.

  • Data Type : string
  • Importance: medium
  • Optional : yes
  • Default : localhost


The password of the key store

  • Data Type : string
  • Importance: medium
  • Optional : yes
  • Default : rootPass


The path to the keystore.

  • Data Type : string
  • Importance: medium
  • Optional : yes
  • Default :


The password of the trust store

  • Data Type : string
  • Importance: medium
  • Optional : yes
  • Default : rootPass


The path to the truststore.

  • Data Type : string
  • Importance: medium
  • Optional : yes
  • Default :


The certificates to load from the trust store.

  • Data Type : list
  • Importance: medium
  • Optional : yes
  • Default :


The key to use to get the certificate chain.

  • Data Type : string
  • Importance: medium
  • Optional : yes
  • Default : client

Schema Evolution

The schema is fixed.

The following schema is used for the key:

Name Type
source Optional string
source_resource Optional String
message_id Optional int32

The following schema is used for the payload:

Name Type
message_id Optional int32
type Optional String
code Optional String
raw_code Optional int32
rtt Optional int64
is_last Optional boolean
is_notification Optional boolean
source Optional String
destination Optional String
timestamp Optional int64
token Optional String
is_duplicate Optional boolean
is_confirmable Optional boolean
is_rejected Optional boolean
is_acknowledged Optional boolean
is_canceled Optional boolean
accept Optional int32
block1 Optional String
block2 Optional String
content_format Optional int32
etags Array of Optional Strings
location_path Optional String
location_query Optional String
max_age Optional int64
observe Optional int32
proxy_uri Optional String
size_1 Optional String
size_2 Optional String
uri_host Optional String
uri_port Optional int32
uri_path Optional String
uri_query Optional String
payload Optional String