Partner Development Guide for
Kafka Connect

Overview
This guide is intended to provide useful background to developers implementing Kafka Connect
sources and sinks for their data stores.

Last Updated: December 2016

© 2014–2016 Confluent, Inc.

December 2016

Getting Started ............................................................................................................................. 2
Community Documentation (for basic background) ................................................................................... 2
Kafka Connect Video Resources (for high-level overview)........................................................................ 2
Sample Connectors .................................................................................................................................... 2
Developer Blog Post (for a concrete example of end-to-end design) ........................................................ 2

Developing a Certified Connector: The Basics ........................................................................ 3
Coding ......................................................................................................................................................... 3
Documentation and Licensing .................................................................................................................... 4
Unit Tests .................................................................................................................................................... 4
System Tests .............................................................................................................................................. 5
Packaging ................................................................................................................................................... 6

Development Best Practices for Certified Connectors ........................................................... 7
Connector Configuration ............................................................................................................................. 7
Schemas and Schema Migration................................................................................................................ 8Type support ........................................................................................................................................... 8
Logical Types .......................................................................................................................................... 8
Schemaless data ..................................................................................................................................... 9
Schema Migration ................................................................................................................................... 9
Offset Management .................................................................................................................................. 10
Source Connectors ............................................................................................................................... 10
Sink Connectors .................................................................................................................................... 11
Converters and Serialization .................................................................................................................... 11
Parallelism ................................................................................................................................................. 12
Error Handling ........................................................................................................................................... 13

Connector Certification Process ............................................................................................. 14
Post-Certification Evangelism .................................................................................................. 14© 2014–2016 Confluent, Inc.

| 1

December 2016

Getting Started
Community Documentation (for basic background)Kafka Connect Overview

Kafka Connect Developer's Guide

Kafka Connect Video Resources (for high-level overview)
A broad set of video resources is available at https://vimeo.com/channels/1075932/videos . Of
particular interest to Connector developers are:
1. Partner Technology Briefings : Developing Connectors in the Kafka Connect Framework
2. Partner Tech Deep Dive : Kafka Connect Overview( https://drive.google.com/open?id=0B6lWkg0jB5xlOUR6NHdrYUwyNmM )3. Partner Tech Deep Dive : Kafka Connect Sources and Sinks( https://drive.google.com/open?id=0B6lWkg0jB5xlV0xoWlZEZjFDdkE )Sample Connectors
1. Simple file connectors: https://github.com/confluentinc/kafka/tree/trunk/connect/file/src/main/java/org/apache/kafka/connect/file
2. JDBC Source/Sink : https://github.com/confluentinc/kafka-connect-jdbc
3. Cassandra (DataStax Enterprise) : https://github.com/datamountaineer/streamreactor/tree/master/kafka-connect-cassandra

Other supported and certified connectors are available at http://confluent.io/product/connectors .

Developer Blog Post (for a concrete example of end-to-end design)
See <Jeremy Custenboarder's Blog Post>(currently https://gist.github.com/jcustenborder/b9b1518cc794e1c1895c3da7abbe9c08)© 2014–2016 Confluent, Inc.

| 2

December 2016

Developing a Certified Connector: The Basics
CodingConnectors are most often developed in Java. While Scala is an acceptable alternative, theincompatibilities between Scala 2.x run-time environments might make it necessary to distributemultiple builds of your connector. Java 8 is recommended.

Confluent engineers have developed a Maven archetype to generate the core structure of yourconnector.

mvn archetype:generate -B -DarchetypeGroupId=io.confluent.maven \-DarchetypeArtifactId=kafka-connect-quickstart \-DarchetypeVersion=0.10.0.0 \-Dpackage=com.mycompany.examples \-DgroupId=com.mycompany.examples \-DartifactId=testconnect \-Dversion=1.0-SNAPSHOT
will create the source-code skeleton automatically, or you can select the options interactively with

mvn archetype:generate -DarchetypeGroupId=io.confluent.maven \-DarchetypeArtifactId=kafka-connect-quickstart \-DarchetypeVersion=0.10.0.0
This archetype will generate a directory containing Java source files with class definitions and stubfunctions for both Source and Sink connectors. You can choose to remove one or the othercomponents should you desire a uni-directional connector. The archetype will also generate some
simple unit-test frameworks that should be customized for your connector.
Note on Class Naming : The Confluent Control Center supports interactive configuration ofConnectors (see notes on Connector Configuration below). The naming convention that allows
Control Center to differentiate sources and sinks is the use of SourceConnector and
SinkConnector as Java classname suffixes (eg. JdbcSourceConnector
and JdbcSinkConnector). Failure to use these suffixes will prevent Control Center fromsupporting interactive configuration of your connector.
© 2014–2016 Confluent, Inc.

| 3

December 2016

Documentation and Licensing
The connector should be well-documented from a development as well as deployment perspective. At
a minimum, the details should includeTop-level README with a simple description of the Connector, including its data model andsupported delivery semantics.

Configuration details (this should be auto-generated via the toRst/toHtml methods for theConfigDef object within the Connector). Many developers include this generation as part of theunit test framework.



OPTIONAL: User-friendly description of the connector, highlighting the more importantconfiguration options and other operational details



Quickstart Guide : end-to-end description of moving data to/from the Connector. Often, thisdescription will leverage the kafka-console-* utilities to serve as the other end of the data
pipeline (or kafka-console-avro-* when the Schema Registry-compatible Avro converter classes
are utilized).

See the JDBC connector for an example of comprehensive Connector documentation
https://github.com/confluentinc/kafka-connect-jdbc

Most connectors will be developed to OpenSource Software standards, though this is not arequirement. The Kafka Connect framework itself is governed by the Apache License, Version
2.0. The licensing model for the connector should be clearly defined in the documentation. Whenapplicable, OSS LICENSE files must be included in the source code repositories.

Unit Tests
The Connector Classes should include unit tests to validate internal API's. In particular, unit tests
should be written for configuration validation, data conversion from Kafka Connect framework to any
data-system-specific types, and framework integration. Tools like PowerMock( https://github.com/jayway/powermock ) can be utilized to facilitate testing of class methods
independent of a running Kafka Connect environment.

© 2014–2016 Confluent, Inc.

| 4

December 2016

System Tests
System tests to confirm core functionality should be developed. Those tests should verify proper
integration with the Kafka Connect framework:

proper instantiation of the Connector within Kafka Connect workers (as evidenced by proper
handling of REST requests to the Connect workers)schema-driven data conversion with both Avro and JSON serialization classes

task restart/rebalance in the event of worker node failure

Advanced system tests would include schema migration, recoverable error events, and performance
characterization. The system tests are responsible for both the data system endpoint and anynecessary seed data:

System tests for a MySQL connector, for example, should deploy a MySQL database instancealong with the client components to seed the instance with data or confirm that data has been
written to the database via the Connector.



System tests should validate the data service itself, independent of Kafka Connect. This canbe a trivial shell test, but definitely confirm that the automated service deployment is functioning
properly so as to avoid confusion should the Connector tests fail.

Ideally, system tests will include stand-alone and distributed mode testingStand-alone mode tests should verify basic connectivity to the data store and core behaviors(data conversion to/from the data source, append/overwrite transfer modes, etc.). Testing ofschemaless and schema'ed data can be done in stand-alone mode as well.



Distributed mode tests should validate rational parallelism as well as proper failure
handling. Developers should document proper behavior of the connector in the event of worker
failure/restart as well as Kafka Cluster failures. If exactly-once delivery semantics aresupported, explicit system testing should be done to confirm proper behavior.

Absolute performance tests are appreciated, but not required.

The Confluent System Test Framework ( https://cwiki.apache.org/confluence/display/KAFKA/tutorial++set+up+and+run+Kafka+system+tests+with+ducktape ) can be leveraged for more advanced systemtests. In particular, the ducktape framework makes tesing of different Kafka failure modes simpler. An
example of a Kafka Connect ducktape test is available here
© 2014–2016 Confluent, Inc.

| 5

December 2016

: https://github.com/apache/kafka/blob/trunk/tests/kafkatest/tests/connect/connect_distributed_test.py#
L356 .

Packaging
The final connector package should have minimal dependences. The default invocation of theConnect Worker JVM's includes the core Apache and Confluent classes from the distribution in
CLASSPATH. The packaged connectors (e.g. HDFS Sink and JDBC Source/Sink) are deployed toshare/java/kafka-connect-* and included in CLASSPATH as well. To avoid Java namespace collisions,
you must not directly include any of the following classes in your connector jar :io.confluent.*

org.apache.kafka.connect.*

In concrete terms, you'll want your package to depend only on the connect-api artifact, and that artifact
should be classified as provided. That will ensure that no potentially conflicting jar's will be included inyour package.

Kafka Connect 0.10.* and earlier does not support CLASSPATH isolation within the JVM's deploying
the connectors. If you Connector conflicts with classes from the packaged connectors, you should
document the conflict and the proper method for isolating your Connector at runtime.

Such isolation

can be accomplished by disabling the packaged connectors completely (renaming the share/java/kafkaconnect-* directories) or developing a customized script to launch your Connect Workers thateliminates those directories from CLASSPATH.

Developers are free to distribute their connector via whatever packaging and installation framework ismost appropriate.

Confluent distributes its software as rpm/deb packages as well as a self-contained

tarball for relocatable deployments. Barring extraordinary circumstances, Connector jars should bemade available in compiled form rather than requiring end customers to build the connector onsite. The helper scripts that launch Kafka Connect workers (connect-standalone and connectdistributed) explicitly add the connector jars to the CLASSPATH. By convention, jar files inshare/java/kafka-connect-* directories are added automatically, so you could document your installation
process to locate your jar files in share/java/kafka-connect-<MyConnector> .

© 2014–2016 Confluent, Inc.

| 6

December 2016

Development Best Practices for Certified Connectors
Connector Configuration
Connector classes must define the config() method, which returns an instance of the ConfigDef classrepresenting the required configuration for the connector. The AbstractConfig class should be used tosimplify the parsing of configuration options. That class supports get* functions to assist in
configuration validation. Complex connectors can choose to extend the AbstractConfig class to delivercustom functionality. The existing JDBCConnector illustrates that with its JDBCConnectorConfig class,which extends AbstractConfig while implementing the getBaseConfig() method to return the necessary
ConfigDef object when queried. You can see how ConfigDef provides a fluent API that lets you easily
define new configurations with their default values and simultaneously configure useful UI parameters
for interactive use. An interesting example of this extensibility can be found in the MODE_CONFIG
property within JDBCConnectorConfig. That property is constrained to one of 4 pre-defined values
and will be automatically validated by the framework.

The ConfigDef class instance within the Connector should handle as many of the configuration details(and validation thereof) as possible. The values from ConfigDef will be exposed to the REST interface
and directly affect the user experience in Confluent Control Center. For that reason, you shouldcarefully consider grouping and ordering information for the different configuration
parameters. Parameters also support Recommender functions for use within the Control Center
environment to guide users with configuration recommendations. The connectors developed by the
Confluent team (JDBC, HDFS, and Elasticsearch) have excellent examples of how to construct a
usable ConfigDef instance with the proper information.

If the configuration parameters are interdependent, implementing a <Connector>.validate() function ishighly recommended. This ensures that the potential configuration is consistent before it is used forConnector deployment. Configuration validation can be done via the REST interface before deploying
a connector; the Confluent Control Center always utilizes that capability so as to avoid invalid
configurations.

© 2014–2016 Confluent, Inc.

| 7

December 2016

Schemas and Schema Migration
Type support
The Connector documentation should, of course, include all the specifics about the data types
supported by your connector and the expected message syntax. Sink Connectors should not simply
cast the fields from the incoming messages to the expected data types. Instead, you should check themessage contents explicitly for your data objects within the Schema portion of the SinkRecord (orwith instanceof for schemaless data). The PreparedStatementBinder.bindRecord() method in the
JdbcSinkConnector provides a good example of this logic. The lowest level loop walks through all the
non-key fields in the SinkRecords and converts those fields to a SQLCompatible type based on theConnect Schema type associated with that field:for (final String fieldName : fieldsMetadata.nonKeyFieldNames) {final Field field = record.valueSchema().field(fieldName);bindField(index++, field.schema().type(), valueStruct.get(field));}
Well-designed Source Connectors will associate explicit data schemas with their messages, enabling
Sink Connectors to more easily utilize incoming data. Utilities within the Connect framework simplify
the construction of those schemas and their addition to the SourceRecords structure.

The code should throw appropriate exceptions if the data type is not supported. Limited data typesupport won't be uncommon (e.g. many table-structured data stores will require a Structwith name/value pairs). If your code throws Java exceptions to report these errors, a best practice is to
use ConnectException rather than the potentially confusing ClassCastException. This will ensure themore useful status reporting to Connect's RESTful interface, and allow the framework to manage yourconnector more completely.

Logical Types
Where possible, preserve the extra semantics specified by logical types by checking
for schema.name()'s that match known logical types. Although logical types will safely fall back on thenative types (e.g. a UNIX timestamp will be preserved as a long), often times systems will provide a
corresponding type that will be more useful to users. This will be particularly true in some common
cases, such as Decimal, where the native type (bytes) does not obviously correspond to the logical© 2014–2016 Confluent, Inc.

| 8

December 2016

type. The use of schema in these cases actually expands the functionality of the connectors ... andthus should be leveraged as much as possible.

Schemaless data
Connect prefers to associate schemas with topics and we encourage you to preserve those schemas
as much as possible. However, you can design a connector that supports schemaless data. Indeed,some message formats implicitly omit schema (eg JSON). You should make a best effort to support
these formats when possible, and fail cleanly and with an explanatory exception message when lack ofa schema prevents proper handling of the messages.

Sink Connectors that support schemaless data should detect the type of the data and translate it
appropriately. The community connector for DynamoDB illustrates this capability very clearly in its
AttributeValueConverter class. If the connected data store requires schemas and doesn't efficiently
handle schema changes, it will likely prove impossible to handle implicit schema changes
automatically. It is better in those circumstances to design a connector that will immediately throw anerror. In concrete terms, if the target data store has a fixed schema for incoming data, by all means
design a connector that translates schemaless data as necessary. However, if schema changes in theincoming data stream are expected to have direct effect in the target data store, you may wish to
enforce explicit schema support from the framework.

Schema Migration
Schemas will change, and your connector should expect this.

Source Connectors won't need much support as the topic schema is defined by the source system; tobe efficient, they may want to cache schema translations between the source system and Connect's
data API, but schema migrations "just work" from the perspective of the framework. Source
Connectors may wish to add some data-system-specific details to their error logging in the event ofschema incompatibility exceptions. For example, users could inadvertently configure two instances of
the JDBC Source connector to publish data for table "FOO" from two different database instances. A
different table structure for FOO in the two databases would result in the second Source Connector
getting an exception when attempting to publish its data ... and error message indicating the specific
JDBC Source would be helpful.
© 2014–2016 Confluent, Inc.

| 9

December 2016

Sink Connectors must consider schema changes more carefully. Schema changes in the input data
might require making schema changes in the target data system before delivering any data with the
new schema. Those changes themselves could fail, or they could require compatible transformations
on the incoming data. Schemas in the Kafka Connect framework include a version. Sink connectors
should keep track of the latest schema version that has arrived. Remember that incoming data may
reference newer or older schemas, since data may be delivered from multiple Kafka partitions with anarbitrary mix of schemas. By keeping track of the Schema version, the connector can ensure that
schema changes that have been applied to the target data system are not reverted. Converters within
the Connector can the version of the schema for the incoming data along with the latest schema
version observed to determine whether to apply schema changes to the data target or to project theincoming data to a compatible format. Projecting data between compatible schema versions can bedone using the SchemaProjector utility included in the Kafka Connect framework. The
SchemaProjector utility leverages the Connect Data API, so it will always support the full range of datatypes and schema structures in Kafka Connect.

Offset Management
Source Connectors
Source Connectors should retrieve the last committed offsets for the configured Kafka topics during theexecution of the start() method. To handle exactly-once semantics for message delivery, the Source
Connector must correctly map the committed offsets to the Kafka cluster with some analog within thesource data system, and then handle the necessary rewinding should messages need to be redelivered. For example, consider a trivial Source connector that publishes the lines from an input file toa Kafka topic one line at a time ... prefixed by the line number. The commit* methods for thatconnector would save the line number of the posted record ... and then pick up at that location upon arestart.

An absolute guarantee of exactly once semantics is not yet possible with Source Connectors (there arevery narrow windows where multiple failures at the Connect Worker and Kafka Broker levels could
distort the offset management functionality). However, the necessary low-level changes to the Kafka
Producer API are being integrated into Apache Kafka Core to eliminate these windows.

© 2014–2016 Confluent, Inc.

| 10

December 2016

Sink Connectors
The proper implementation of the flush() method is often the simplest solution for correct offset
management within Sink Connectors. So long as Sinks correctly ensure that messages delivered tothe put() method before flush() are successfully saved to the target data store before returning from theflush() call, offset management should "just work". A conservative design may choose not to
implement flush() at all and simply manage offsets with every put() call. In practice, that design may
constrain connector performance unnecessarily.
Developers should carefully document and implement their handling of partitioned topics. Since
different connector tasks may receive data from different partitions of the same topic, you may needsome additional data processing to avoid any violations of the ordering semantics of the target datastore. Additionally, data systems where multiple requests can be "in flight" at the same time frommultiple Connector Task threads should make sure that relevant data ordering is preserved (eg notcommitting a later message while an earlier one has yet to be confirmed). Not all target data systems
will require this level of detail, but many will.

Exactly-once semantics within Sink Connectors require atomic transactional semantics against thetarget data system ... where the known topic offset is persisted at the same time as the payload
data. For some systems (notably relational databases), this requirement is simple. Other target
systems require a more complex solution. The Confluent-certified HDFS connector offers a good
example of supporting exactly-once delivery semantics using a connector-managed commit strategy.

Converters and Serialization
Serialization formats for Kafka are expected to be handled separately from connectors. Serialization is
performed after the Connect Data API formatted data is returned by Source Connectors, or before
Connect Data API formatted data is delivered to Sink Connectors. Connectors should not assume aparticular format of data. However, note that Converters only address one half of the system.
Connectors may still choose to implement multiple formats, and even make them pluggable. For
example, the HDFS Sink Connector (taking data from Kafka and storing it to HDFS) does not assume
anything about the serialization format of the data in Kafka. However, since that connector is
responsible for writing the data to HDFS, it can handle converting it to Avro or Parquet, and even allows
users to plug in new format implementations if desired. In other words, Source Connectors might beflexible on the format of source data and Sink Connectors might be flexible on the format of sink data,
but both types of connectors should let the Connect framework handle the format of data within Kafka.
© 2014–2016 Confluent, Inc.

| 11

December 2016

There are currently two supported data converters for Kafka Connect distributed with
Confluent: org.apache.kafka.connect.json.JsonConverter and io.confluent.connect.avro.AvroConverter. Both converters support including the message schema along with the payload (when configured withthe appropriate *.converter.schemas.enable property to true).
The JsonConverter includes the schema details as simply another JSON value in each record. A
record such as " {"name":"Alice","age":38} " would get wrapped to the longer format{"schema":{"type":"struct","fields":[{"type":"string","optional":false,"field":"name"},{"type":"integer","optional":false,"field":"age"}],"optional":false,"name":"htest2"},"payload":{"name":"Alice","age":38}}
Connectors are often tested with the JsonConverter because the standard Kafka consumers andproducers can validate the topic data.

The AvroConverter uses the SchemaRegistry service to store topic schemas, so the volume of data on
the Kafka topic is much reduced. The SchemaRegistry enabled Kafka clients (eg kafka-avro-consoleconsumer) can be used to examine these topics (or publish data to them).

Parallelism
Most connectors will have some sort of inherent parallelism. A connector might process many log files,
many tables, many partitions, etc. Connect can take advantage of this parallelism and automaticallyallow your connector to scale up more effectively – IF you provide the framework the necessary
information.

Sink connectors need to do little to support this because they already leverage Kafka's consumer
groups functionality; recall that consumer groups automatically balance and scale work between
member consumers (in this case Sink Connector tasks) as long as enough Kafka partitions areavailable on the incoming topics.

© 2014–2016 Confluent, Inc.

| 12

December 2016

Source Connectors, in contrast, need to express how their data is partitioned and how the work of
publishing the data can be split across the desired number of tasks for the connector. The first step is todefine your input data set to be broad by default, encompassing as much data as is sensible given asingle configuration. This provides sufficient partitioned data to allow Connect to scale up/downelastically as needed. The second step is to use your Connector.taskConfigs() method implementation
to divide these source partitions among (up to) the requested number of tasks for the connector.
Explicit support of parallelism is not an absolute requirement for Connectors – some data sources
simply do not partition well. However, it is worthwhile to identify the conditions under which parallelism
is possible. For example, a database might have a single WAL file which seems to permit noparallelism for a change-data-capture connector; however, even in this case we might extract subsetsof the data (e.g. per table from a DB WAL) to different topics, in which case we can get someparallelism (split tables across tasks) at the cost of the overhead of reading the WAL multiple times.

Error Handling
The Kafka Connect framework defines its own hierarchy of throwable error classes(https://kafka.apache.org/0100/javadoc/org/apache/kafka/connect/errors/packagesummary.html ). Connector developers should leverage those classes(particularly ConnectException and RetriableException) to standardize connector behavior. Exceptions
caught within your code should be rethrown as connect.errors whenever possible to ensure proper
visibility of the problem outside the framework. Specifically, throwing a RuntimeException beyond thescope of your own code should be avoided because the framework will have no alternative but toterminate the connector completely.

Recoverable errors during normal operation can be reported differently by sources and sinks. Source
Connectors can return null (or an empty list of SourceRecords) from the poll() call. Those connectors
should implement a reasonable backoff model to avoid wasteful Connector operations; a simple call tosleep() will often suffice. Sink Connectors may throw a RetriableException from the put() call in theevent that a subsequent attempt to store the SinkRecords is likely to succeed. The backoff period forthat subsequent put() call is specified by the timeout value in the sinkContext. A default timeout value
is often included with the connector configuration, or a customized value can be assigned
using sinkContext.timeout() before the exception is thrown.

© 2014–2016 Confluent, Inc.

| 13

December 2016

Connectors that deploy multiple threads should use context.raiseError() to ensure that the framework
maintains the proper state for the Connector as a whole. This also ensures that the exception ishandled in a thread-safe manner.

Connector Certification Process
Partners will provide the following material to the Confluent Partner team for review prior to certification.
1. Engineering materials
a. Source code details (usually a reference to a public source-code repository)b. Results of unit and system tests
2. Connector Hub details
a. Tags / description
b. Public links to source and binary distributions of the connectorc. Landing page for the connector (optional)
3. Customer-facing materials
a. Connector documentation
b. [Recommended] Blog post and walk-through video (eg.Cassandra Sink Blog
at http://www.confluent.io/blog/kafka-connect-cassandra-sink-the-perfect-match/ )The Confluent team will review the materials and provide feedback. The review process may be
iterative, requiring minor updates to connector code and/or documentation. The ultimate goal is acustomer-ready deliverable that exemplifies the best of the partner product and Confluent.

Post-Certification Evangelism
Confluent is happy to support Connect Partners in evangelizing their work. Activities include


Blog posts and social media amplificationCommunity education (meet-ups, webinars, conference presentations, etc)

Press Releases (on occasion)



Cross-training of field sales teams© 2014–2016 Confluent, Inc.

| 14