Enrico Olivelli

Andrey Yegorov

<p>Large-scale implementations of any system, such as the event-streaming platform Apache Kafka, often involve customizations and tools and plugins developed in-house. When it&rsquo;s time to transition from one system to another, the task can become complicated, drawn-out, and error-prone. Often the benefits of an alternative system (which can include <a href="https://gigaom.com/report/the-cost-savings-of-replacing-kafka-with-pulsar/">significant cost savings</a> and other efficiencies) are outweighed by the risks and costs of migration. As a result, an organization can end up locked into a suboptimal situation, footing a bigger bill than necessary and missing out on modern features that help move the business forward faster.&nbsp;</p>
<p>These risks and costs can be mitigated by making the transition process iterative, breaking off the vendor lock-in in small, manageable steps, and avoiding the "big bang" switch that often results in delay delivery and increase the cost of running two systems in parallel for A|B testing.&nbsp;</p>
<p>Let's take a quick look at the existing ecosystem that helps navigate the transition from Kafka to Apache Pulsar, dive into the new addition to the ecosystem in Pulsar 2.8, and look at important changes in Pulsar IO API and Pulsar Schema API that automate and simplify schema handling in Sinks.</p>
<p>Throughout this article, we will follow the convention of discussing streaming <em>sources</em>, that push data into Pulsar from another system, and <em>sinks</em>, that send data from Pulsar to another destination.</p>
<h2>Current state of the Pulsar-Kafka ecosystem</h2>
<h3>Built-in connectors to Kafka</h3>
<p>Built-in connectors simplify pulling/pushing data between Pulsar and Kafka topics.&nbsp;</p>
<p>This is useful if you want to leave existing systems running on Kafka while building new functionality on Pulsar.</p>
<p>More details are available in the Pulsar documentation:</p>
<ul>
<li>Source (ingest data to Pulsar from Kafka): <a href="https://pulsar.apache.org/docs/en/io-kafka-source/">https://pulsar.apache.org/docs/en/io-kafka-source/</a>&nbsp;</li>
<li>Sink (send data from Pulsar to Kafka): <a href="https://pulsar.apache.org/docs/en/io-kafka-sink/">https://pulsar.apache.org/docs/en/io-kafka-sink/</a></li>
</ul>
<h3>Kafka on Pulsar</h3>
<p>Kafka on Pulsar (KoP) is the recommended way to use the native Kafka client with Pulsar.&nbsp;</p>
<p>KoP is a <em>protocol handler</em>. This means that it interprets the Kafka protocol at the network level and translates it into Pulsar requests. There are three key advantages to this approach:</p>
<ol>
<li>KoP works with all Kafka clients</li>
<li>KoP uses the well-defined interface between Kafka client and server</li>
<li>Client code does not need to change at all</li>
</ol>
<h3>Kafka Connect Adaptor</h3>
<p>Most people use Kafka (and Pulsar) via connectors to other systems, rather than writing low-level client code by hand. Pulsar has <a href="https://pulsar.apache.org/docs/en/io-overview/">native connectors</a> available for the most popular systems, but as of this writing, there are many more connectors for Kafka that do not yet exist for Pulsar, including private connectors created in-house for use at a single company.</p>
<p>The Kafka Connect Adaptor (KCA) bridges this gap. KCA is a Pulsar Source and Sink that runs a Kafka Connect Sink or Source. The Kafka Connect Adaptor Sink is new in Pulsar 2.8.</p>
<p>Currently, the documentation is scarce, but using KCA is simple. We will look at examples of using both the KCA Sink and Source below.</p>
<h2>Using the Kafka Connect Adaptor Sink</h2>
<p>Using Kafka Connect Adaptor Sink is fairly straightforward. All you need to do is package the Kafka Connect connector, create the configuration, and use it as a regular Pulsar Sink.</p>
<h3>Step 1: Package</h3>
<p>Use Kafka Connect Adaptor NAR <a href="https://github.com/apache/pulsar/tree/master/pulsar-io/kafka-connect-adaptor-nar">https://github.com/apache/pulsar/tree/master/pulsar-io/kafka-connect-adaptor-nar</a> as a starting point (for simplicity, I'll edit it directly) and add your Kafka Connector Sink to the list of the dependencies in pom.xml. Here&rsquo;s what this would look like with the Kinesis Kafka connector sink:</p>
<div class="code">
<div class="code__heading">Diff<button class="code__copy">Copy</button></div>
<div class="code__body">
<pre><code class="language-ini">
diff --git a/pulsar-io/kafka-connect-adaptor-nar/pom.xml b/pulsar-io/kafka-connect-adaptor-nar/pom.xml
index ea9bedbd056..c7fa9a1ebca 100644
--- a/pulsar-io/kafka-connect-adaptor-nar/pom.xml
+++ b/pulsar-io/kafka-connect-adaptor-nar/pom.xml
@@ -36,6 +36,11 @@
       pulsar-io-kafka-connect-adaptor
       ${project.version}
     
+ 
+ com.amazonaws
+ amazon-kinesis-kafka-connector
+ 0.0.9-SNAPSHOT
+ 
 
      </code></pre>
</div>
</div>
<p><br />Build the NAR:<br />&nbsp;</p>
<pre>$ mvn -f pulsar-io/kafka-connect-adaptor-nar/pom.xml clean package -DskipTests</pre>
<p>&nbsp;</p>
<h3>Step 2: Configuration</h3>
<p>The Sink expects "processingGuarantees" to be "EFFECTIVELY_ONCE"`, configs pointing to the Pulsar instance &amp; topic to store processed offsets at, topic to read the data from, and configuration to pass to the Kafka Connect Sink.</p>
<p>For example:</p>
<div class="code">
<div class="code__heading">YAML<button class="code__copy">Copy</button></div>
<div class="code__body">
<pre><code class="language-ini">
processingGuarantees: "EFFECTIVELY_ONCE"
configs:
  "topic": "my-topic"
  "offsetStorageTopic": "kafka-connect-sink-offset-kinesis"
  "pulsarServiceUrl": "pulsar://localhost:6650/" 
  "kafkaConnectorSinkClass": "com.amazon.kinesis.kafka.AmazonKinesisSinkConnector"
  # The following properties passed directly to Kafka Connect Sink and defined by it
  "kafkaConnectorConfigProperties":
     "name": "test-kinesis-sink"
     'connector.class': "com.amazon.kinesis.kafka.AmazonKinesisSinkConnector"
     "tasks.max": "1"
     "topics": "my-topic"
     "kinesisEndpoint": "kinesis.us-east-1.amazonaws.com"
     "region": "us-east-1"
     "streamName": "test-kinesis"
     "singleKinesisProducerPerPartition": "true"
     "pauseConsumption": "true"
     "maxConnections": "1"
      </code></pre>
</div>
</div>
<h3>Step 3: Profit!</h3>
<p>Follow regular Pulsar's steps to use the packaged connector: <a href="https://pulsar.apache.org/docs/en/io-use/">https://pulsar.apache.org/docs/en/io-use/</a>&nbsp;</p>
<h2>Using the Kafka Connect Adaptor Source</h2>
<p>KCA Source has been available since Pulsar version 2.3.0. In the simplest case, its usage is similar to the KCA Sink&rsquo;s: add the dependency and build, provide configuration and run.</p>
<p>Currently, KCA Source only supports Sources that return data in Apache Avro or JSON formats.&nbsp;</p>
<p>For detailed examples of the use of the Source Adaptor please look at <a href="https://github.com/apache/pulsar/blob/master/pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/DebeziumSource.java">Pulsar&rsquo;s Debezium Connector</a>.</p>
<h2>Under the hood: Building a better developer experience for Pulsar IO</h2>
<p>Apache Pulsar 2.8 offers many improvements to the Java Pulsar Schema API and to the Pulsar IO API that helped to fill in the gaps between Kafka Connect and Pulsar IO. These improvements were foundational for Kafka Connect Adaptor Sink and result in easier development of Pulsar IO Sinks in general.</p>
<p>The Kafka Connect user must explicitly configure the Sink (or the Kafka Consumer) deserializer configuration in order to use the correct deserializer, even if the code is not tied to a particular schema. The power of the updated Pulsar Schema API makes everything automatic and removes the need for explicit configurations.</p>
<p>Let&rsquo;s take a deeper look at the Pulsar IO API improvements below; for more technical details, please refer to the <a href="https://docs.google.com/document/d/1VWi5LHP44V31nP4bCui9d5RXwH6xc_phrUes6tvNguk/edit?usp=sharing">PIP-85</a>.</p>
<h3>Runtime handling of the schema</h3>
<p>We have contributed the support for coding schema-aware Pulsar IO Sinks that do not depend on a particular schema at build time. In other words, in Pulsar 2.7 you had to declare the schema type in your sink:</p>
<div class="code">
<div class="code__heading">Java<button class="code__copy">Copy</button></div>
<div class="code__body">
<pre><code class="language-ini">
class MySink implements Sink {
     public void write(Record record) {
     }
}
      </code></pre>
</div>
</div>
<p>To support &ldquo;String&rdquo; and &ldquo;GenericRecord&rdquo; (JSON and Avro structures) you had to create two classes and the user who deploys the Sink had to use the &ldquo;--classname&rdquo; argument to set the correct implementation for the given topic.</p>
<p>In Pulsar 2.8 you can simply use this syntax:</p>
<div class="code">
<div class="code__heading">Java<button class="code__copy">Copy</button></div>
<div class="code__body">
<pre><code class="language-ini">
class MySink implements Sink {
      public void write(Record record) {}
}
      </code></pre>
</div>
</div>
<p>This sink will work with every schema type and with topics without a schema. It also supports schema evolution and KeyValue schema type.</p>
<h3>Seamless support of KeyValue messages</h3>
<p>The second gap between Kafka Connect and Pulsar IO was the lack of seamless support for KeyValue messages.</p>
<p>For many versions, Pulsar offered the powerful KeyValue schema type that supported setting a schema for the Key and the Value. With a Sink&lt;GenericObject&gt; you can handle the KeyValue schema as well, writing your code only once and keeping it simpler.</p>
<h3>Access message and schema details for messages consumed with Schema.AUTO_CONSUME</h3>
<p>Pulsar uses a special AUTO_CONSUME schema to validate and deserialize messages using schemas received from the broker. Currently, it supports Avro, JSON, and ProtobufNativeSchema schemas. You can find more details in the documentation <a href="https://pulsar.apache.org/docs/en/schema-understand/#auto_consume">https://pulsar.apache.org/docs/en/schema-understand/#auto_consume</a>&nbsp;</p>
<p>Before Pulsar 2.8, AUTO_CONSUME allowed you to decode the message according to the version of the schema attached to the message but did not allow access to the exact schema definition. Pulsar 2.8 enhances the API by providing access to this information:</p>
<div class="code">
<div class="code__heading">Java<button class="code__copy">Copy</button></div>
<div class="code__body">
<pre><code class="language-ini">
Schema schema = message.getReaderSchema().get();

org.apache.avro.Schema avroSchema = (org.apache.avro.Schema) schema.getNativeSchema().get();
org.apache.avro.generic.GenericRecord nativeRecord = (org.apache.avro.generic.GenericRecord) consumedRecord.getNativeObject();
      </code></pre>
</div>
</div>
<p>Message.getReaderSchema() method returns the actual schema used for decoding the message, even in the case of the special AUTO_CONSUME Schema. Such schema automatically downloads new versions of the Schema while the topic evolves.&nbsp;</p>
<p>Schema.getNativeSchema() and GenericRecord.getNativeObject() methods provide access to the underlying implementation of the schema and the Java model of the message. In particular, you can access the Avro schema and the Avro GenericObject instance under the covers.<br />&nbsp;</p>
<h2>Summing up</h2>
<p>The new Kafka Connect Adaptor completes the Pulsar-Kafka compatibility ecosystem. This ecosystem currently allows an iterative transition from Kafka to Pulsar, supports the use of native Kafka clients with Pulsar, the use of Kafka Connect connectors on Pulsar, and data transfer between two systems.&nbsp;</p>
<p>With all these great features available, we hope focus shifts from worrying about the complexity of onboarding Pulsar over existing Kafka implementations to finding new ways their business can benefit from the power of Pulsar.&nbsp;</p>
<p><strong>Want to try out Apache Pulsar?&nbsp;</strong><a href="https://astra.datastax.com/register">Sign up now</a> for <a href="https://www.datastax.com/products/astra-streaming">Astra Streaming</a>, our fully managed Apache Pulsar service. We&rsquo;ll give you access to its full capabilities entirely free through beta. See for yourself how easy it is to build modern data applications and let us know what you&rsquo;d like to see to make your experience even better.&nbsp;</p>
<p>Want to learn more? <a href="https://www.datastax.com/resources/webinar/how-apache-pulsar-helps-overstock-win-in-ecommerce">Sign up for our webinar</a> with Overstock.com on June 29 at 9 am PT to hear how the leading internet retailer wins in ecommerce with Apache Pulsar.</p>

Simplify migrating from Kafka to Pulsar with Kafka Connect Support

Enrico Olivelli

Andrey Yegorov

Share

More Technology

Getting Started with DataStax Astra DB and Amazon Bedrock

Why You Can't Miss Cassandra Summit

Mitigate the OpenAI Debacle with Open Source and Cloud

Supercharge AI Assistants with Superagent and DataStax

NoSQL and Vector DB
for Generative AI,
Instantly, At Scale

Enrico Olivelli

Andrey Yegorov

Share

More Technology

Getting Started with DataStax Astra DB and Amazon Bedrock

Why You Can't Miss Cassandra Summit

Mitigate the OpenAI Debacle with Open Source and Cloud

Supercharge AI Assistants with Superagent and DataStax

NoSQL and Vector DBfor Generative AI,Instantly, At Scale

NoSQL and Vector DB
for Generative AI,
Instantly, At Scale