DEV Community: Neil Buesing

Apache Kafka without Zookeeper and Dedicated Controllers

Neil Buesing — Thu, 26 Jan 2023 04:17:33 +0000

Are you interested in setting up Kafka without Zookeeper and with a dedicated controller quorum? Here are the steps and reference project showcasing how to do this using the Confluent community-licensed container images. A Grafana dashboard to observe the new metrics is also provided.

Introduction

Kafka Raft brings the consensus protocol into the Controller Plane of Apache Kafka from Zookeeper. With this change, the role of a Kafka instance can be that of a controller, broker, or both. Getting this configuration stood up requires some tweaks to the confluent cp-kafka image.

If you want a deeper understanding of the design and implementation details, check out Jun Rao's course on Kafka Internals. Specifically, the control-plane section.

Configuration

There are many configuration parameters with Apache Kafka; highlighted here are the ones necessary to build out the cluster with KRaft.

node.id

The property node.id replaces broker.id. Be sure that all identifiers in the cluster are unique across brokers and controllers.

process.roles

A node can be both a broker or a controller. Set to broker,controller to enable both the controller and data planes on a node.

controller.listener.names

List out the listener names used for the controller. This indicates to a node the listener to use for that communication. While the property is a list, just like advertised.listeners, the first one is what is used for controller communication.

controller.quorum.voters

A comma delimited list of voters in the control plane, where a controller is noted as: node_id@hostname:port.

example:

KAFKA_CONTROLLER_QUORUM_VOTERS: 10@controller-0:9093,11@controller-1:9093,12@controller-2:9093

Additional Configuration

There are other tuning parameters for the controller plan, see the documentation for details.

Lesson Learned

Do not remove cluster settings from the dedicated controllers, since a controller is the node that performs administration operations, such as creating a topic.

Incorrectly removing these from the controllers caused topics to be created without Apache Kafka defaults.

KAFKA_DEFAULT_REPLICATION_FACTOR: 3
KAFKA_NUM_PARTITIONS: 4

Storage

Another change to setting up Apache Kafka with Raft is the storage. The storage on each node must be configured, before starting the JVM. This can be done with a kafka-storage command provided as part of Apache Kafka.

Generate a unique UUID for the cluster, you can use kafka-storage random-uuid or another means.
Before starting the cluster, format the metadata storage with kafka-storage format.

kafka-storage format -t $KAFKA_CLUSTER_ID -c <server.properties>

Container Images

With these details in mind, applying them to Confluent's cp-kafka image takes a little finesse, at least with version 7.3.0. The cp-kafka container's entry point, /etc/confluent/docker/run, builds the configuration for Apache Kafka from environment variables following conventions. In addition, there are validation steps to catch misconfiguration. These validations, however, need to change, since certain assumptions no longer apply in a zookeeper-less setup. In addition, a node that is only for a controller does not define advertised.listeners so validation for that needs to be removed.

Script Modifications

The following tasks need to be done to start up these images with raft.

Remove KAFKA_ZOOKEEPER_CONNECT validation for all nodes.
Remove checking for zookeeper ready state for all nodes.
Remove KAFKA_ADVERTISED_LISTENERS validation for dedicated controller nodes.
Create Metadata store for all nodes by running kafka-storage format.

Command

The cp-kafka image's command is /etc/confluent/docker/run, and the scripts and docker-compose command setting, allow
these containers to start with raft consensus protocol.

broker.sh

volumes:
  - ./broker.sh:/tmp/broker.sh
command: bash -c '/tmp/broker.sh && /etc/confluent/docker/run'

controller.sh

volumes:
  - ./controller.sh:/tmp/controller.sh
command: bash -c '/tmp/controller.sh && /etc/confluent/docker/run'

Seeing It In Action

If you are interested in seeing all this in action, check out the kafka-raft docker-compose
setup in the dev-local project. It is a fully working examples with 3 controllers and 4 brokers.

Metrics

If you are going to deploy Kafka with Raft to production, having visibility to metrics is important. Adding that visibility is just as important (if not more so) than having dedicated controllers. The key to dashboards, ensure that they report correctly on data and control planes are separated or combined.

Grafana

A Grafana Dashboard is a multi-step process of extracting the metrics and storing them in a time-series database (e.g. Prometheus) and then visualizing those collected metrics in a Grafana Dashboard.

JMX Prometheus Exporter

The KRaft Monitor metrics are defined in the documentation, with an MBean name, such as kafka.server:type=raft-metrics,name=current-state. Using a JMX client, such as jmxterm, shows the MBean is name is just kafka.server:type=raft-metrics with each metric an attribute in that bean. This is different from the current documentation.

If you deploy Java applications with JMX Metrics in containers, I highly recommend jmxterm.

java -jar jmxterm.jar

In the cp containers, the java process is process 1, but use the command jvms to see all available processes; and verify the JVM's process-id is indeed 1.

$>open 1

Show all the MBeans in the JVM, with beans.

$>beans
...
kafka.server:type=raft-metrics
...

Select a bean and use info to explore the attributes on a bean and get to get the current value of an attribute.

$>bean kafka.server:type=raft-metrics
#bean is set to kafka.server:type=raft-metrics

$>info
# attributes
%0   - append-records-rate (double, r)
%1   - commit-latency-avg (double, r)
%2   - commit-latency-max (double, r)
%3   - current-epoch (double, r)
%4   - current-leader (double, r)
%5   - current-state (double, r)
%6   - current-vote (double, r)
%7   - election-latency-avg (double, r)
%8   - election-latency-max (double, r)
%9   - fetch-records-rate (double, r)
%10  - high-watermark (double, r)
%11  - log-end-epoch (double, r)
%12  - log-end-offset (double, r)
%13  - number-unknown-voter-connections (double, r)
%14  - poll-idle-ratio-avg (double, r)

$>get current-leader
current-leader = 10.0;

Leveraging the above, the following properly exposes these from JMX Prometheus Exporter. Since current-state attribute value is a string, its value needs to be associated with a label to capture it in Prometheus.

rules:
- pattern: "kafka.server<type=raft-metrics><>(current-state): (.+)"
  name: kafka_server_raft_metrics
  labels:
  name: $1
  state: $2
  value: 1
- pattern: "kafka.server<type=raft-metrics><>(.+): (.+)"
  name: kafka_server_raft_metrics
  labels:
  name: $1

Grafana Dashboard

Grafana is great for custom configuration, but that means time and effort are needed to build them. Here is a dashboard around some of those raft metrics; it is not a complete dashboard.

Node Information

So with metrics emitting, put them into a Grafana dashboard.

By using the current-state we can see who is the leader, in addition to capturing node.id. In addition, the dashboard component joins in other data to provide additional information on each node.

Node Counts

Counts of nodes and active controller are always re-assuring, and this leverages an existing metric, kafka_controller_kafkacontroller_value{name="ActiveControllerCount",}. This metric only is emitted from a controller, so by counting the existence of this metric you see the number of controllers in the cluster, and by summing the value of the metric you get the actual number of active controllers; alert if this is ever not equal to 1.

Check out the dashboard to see how the other values are calculated, as it is the same as in the zookeeper-based installations.

Active Controller

To get the node.id of the quorum leader, just find max(kafka_server_raft_metrics{name="current-leader",}). Because scraping of each node is from slightly different times, different values are possible at the time of a change; max is used to make the single value display easy to build.

If a new leader is being voted, that will show up in the voted leader metric. In a single value dashboard, I do not expect this to be that useful, but in a time-series history of values, more value would be in having this metric recorded.

Full Dashboard

Here is an example dashboard that also captures the fetch rate of the controller metadata.

Open-Source Tools

I have checked a variety of open-source tools out there and have been unsuccessful in seeing raft metrics. Active controller information is incorrectly displayed. None of the tools tried came to support raft, but it is important that before you upgrade, you have a proper monitoring and alerting strategy in place.

Summary

Be sure you properly test your monitoring and Apache Kafka support infrastructure as part of your move to Kafka Raft Consensus Protocol. Also, validate that kafka-raft metrics are captured and ensure that dashboards and tools work when brokers and controllers are on separate nodes.

Console Consumer - BytesDeserializer for the Win

Neil Buesing — Thu, 15 Sep 2022 19:10:57 +0000

If you want to make sure your expected String key is what you think it is, using BytesDeserializer with your console consumers is better than StringDeserializer.

Introduction

The Confluent Avro Serializer and Deserializer leverages storing the unique ID of the schema in the message. When unexpected characters show up in a string, a type mismatch would be more obvious. But what about non-printable characters? How do they show up? Will the issue then be obvious?

Demonstration

A simple demonstration can be done with the Datagen Source Connector. Create a connector with Avro as the key. The data type for the Datagen's quickstart users is a string. The Avro serializer will write this as an Avro primitive. Typically, when Avro is used, the top-level object is a Record, but the serializer has custom code for supporting primitives.

The Configuration

The Datagen connector is configured with the key being represented at Avro.

{
    "connector.class": "io.confluent.kafka.connect.datagen.DatagenConnector",
    "tasks.max": "1",
    "kafka.topic": "users",
    "quickstart": "users",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url" : "http://schema-registry:8081",
    "key.converter.schemas.enable": "true",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url" : "http://schema-registry:8081",
    "value.converter.schemas.enable": "true",
    "max.interval": 100,
    "iterations": 10000000
}

Scenario

You write a Kafka Streams application where you read the key as a Serdes.String(), the default you used for your application. You forget to change the serde for reading users from the default serde to an Avro Serde. You now join your stream of orders with users, and none of the joins succeeds.

Investigation...

If you are me, the first thing you do is you use kafka-avro-console-consumer to see what is going on.

kafka-avro-console-consumer \
        --bootstrap-server localhost:19092 \
        --property schema.registry.url="http://localhost:8081" \
        --property print.key=true \
        --property key.separator="|" \
        --from-beginning \
        --skip-message-on-error \
        --key-deserializer=org.apache.kafka.common.serialization.StringDeserializer \
        --topic users

The result has content that looks pretty normal and expected:

User_9|{"registertime":1489457902486,"userid":"User_9","regionid":"Region_1","gender":"OTHER"}
User_1|{"registertime":1500277798184,"userid":"User_1","regionid":"Region_2","gender":"OTHER"}

Now there could be extra blank lines show up if the non-printable bytes triggers that; but that doesn't always stick out and an obvious issue (at least not obvious to me).

What if your key deserializer was BytesDeserializer, what would you have seen?

kafka-avro-console-consumer \
        --bootstrap-server localhost:19092 \
        --property schema.registry.url="http://localhost:8081" \
        --property print.key=true \
        --property key.separator="|" \
        --from-beginning \
        --skip-message-on-error \
        --key-deserializer=org.apache.kafka.common.serialization.BytesDeserializer \
        --topic users

The serializer's magic byte (0x00) and the bytes for the schema-id show up in printable hex characters:

\x00\x00\x00\x00\x03\x0CUser_9|{"registertime":1489457902486,"userid":"User_9","regionid":"Region_1","gender":"OTHER"}
\x00\x00\x00\x00\x03\x0CUser_1|{"registertime":1500277798184,"userid":"User_1","regionid":"Region_2","gender":"OTHER"}

Now it is easy to see the issue, the key is Avro (a primitive Avro string as defined by the serializer). Solution: update the connector to use a String, or update the streams application to re-key the data.

NOTE

Running containers for demonstrations is great, but the mis-match of URLs can be confusing. localhost:port is used for connecting to services from the host machine (your laptop) via port mapping. The actual hostname is used when you are accessing the service from another container. Therefore, you will see http://schema-registry:8081 within the connect configuration, and http://localhost:8081 for commands running from the host machine. I do not translate here as these scripts align with the demo code.

Useful Shell Aliases

I have these defined in my .zshrc.

alias kcc='kafka-console-consumer \
        --bootstrap-server localhost:19092 \
        --key-deserializer=org.apache.kafka.common.serialization.BytesDeserializer  \
        --property print.key=true \
        --property key.separator="|" \
        --from-beginning \
        --topic'

alias kacc='kafka-avro-console-consumer \
        --bootstrap-server localhost:19092 \
        --property schema.registry.url="http://localhost:8081" \
        --property print.key=true \
        --property key.separator="|" \
        --from-beginning \
        --skip-message-on-error \
        --key-deserializer=org.apache.kafka.common.serialization.BytesDeserializer \
        --topic'

Takeaways

While this may seem obvious to you, and you would immediately inspect the connector configuration and uncover the problem; you want to make things easy for everyone on your team. Allowing them to troubleshoot and find issues easy is a win for you and a win for them.
This demonstration is available within the key-mismatch demo within dev-local-demos.

Reach out

I would enjoy hearing more about the development improvements you use.
Reach out at contact us.

Apache Kafka Monitoring and Management

Neil Buesing — Thu, 08 Sep 2022 16:33:03 +0000

Setting up multiple kafka cluster configurations to explore the nuances of various tools that monitor Apache Kafka.

tl;dr

I have a new project I have created for testing Kafka tools, check it on GitHub at kineticedge/kafka-toolage. It has 4 different Kafka clusters. As I add new tools, I will be adding those configurations to this project and then write about that experience and talk about their configuration and their functionality. I spend a lot of time doing web-searches and trial and error getting configurations working; I hope I can save you some time.

Introduction

When it comes to the configuration of Apache Kafka, the integration can get complicated. The number of configurations of Apache Kafka, Kafka Connect, Confluent Schema Registry is extensive. When you add in 3rd-party tools each built by independent developer teams, it's challenging to figure out all the configurations.

When you want to evaluate a tool, you want to focus on exploring its features and determining if those features meet your needs; you do not want to worry about setup and especially don’t want to evaluate only to find out you cannot easily integrate it with your production environment.

Future articles walk through the setup of tools against these 4 different Apache Kafka cluster configurations. This will accelerate your integration time and leave your developers focused on writing Kafka applications and minimizing time spent on infrastructure configuration.

In addition to seeing how to configure open-source tooling on clusters of each configuration, here are some additional items to see through this journey:

If you want to use a JAAS configuration file, why does the Confluent Container of Schema Registry require you to add it to both KAFKA_OPTS and SCHEMA_REGISTRY_OPTS?
Basic Auth configuration for Schema Registry and Kafka Connect are not the same, what are the differences exactly?
Why is Confluent Schema Registry configuration in Kafka-UI different from Kafka Connect and Confluent’s Java Serializer and Deserializer configurations?

Why Multiple Clusters?

The 4 clusters showcase 5 different Apache Kafka protocols and multiple options for Kafka Connect and Schema Registry.

The non-authenticated cluster is typically like an internal developer cluster or a POC cluster. It is typically the environment used when trying out Kafka for the first time, or an environment where someone is learning more about Kafka. The demonstrated here has two listeners, PLAINTEXT and SSL, but typically if SSL is desired, the PLAINTEXT listener is omitted.

A SASL authenticated cluster gives the ability to use PLAINTEXT and SSL connections. Why would anyone want to do this? Well, if you have internal IPs only accessible from in-network, you have the option for specific tools and connections to leverage 0 page-copy and improve performance of inner broker communication. You also can leverage PLAIN authentication vs. SCRAM authentication for setting up super-user access. Using SCRAM based users for admin users is a little more complicated in that you have to create those users when zookeeper is up and available (before the brokers are ever started) and leverage tooling directly against zookeeper. This gives insight into how to do that.

A SSL authenticated cluster shows how you can leverage certificates for client authentication. Building up a self-signed certificate process is not something that is part of most distributions; this project provides one to showcase how to leverage openssl to create certificates while showcasing the proper X509 extensions necessary.

Finally, an OAUTH authenticate cluster showcases how extensible a tool is for integration into your environment; and will help you identify if multiple authentication means to your cluster is necessary for leveraging a particular tool.

All certificates are all internally created by this project. A single turnkey generation from CA to broker and client certificates; including intermediate CA certificates.

The goal here is to verify that the open-source tool being considered will work with your environment. While the clusters presented here, do not handle all scenarios, it covers many of them and I hope that can minimize the path to getting them integrated with your clusters. I have spent hours on incorrect understandings of configuration, certificates that can be used for client authentication, and the unique configuration settings of many 3rd party tools.

While reading the documentation is typically the answer, juggling many documents when your interest is to verify something meets your needs before you spend all those hours getting it integrated is not ideal; let us help you with that by showcasing it here.

Years of Apache Kafka experience is behind this setup.
Such setups can easily become 2-4 months of effort within your organization. This project's goal is to help you get up and running in a fraction of that time.

The Clusters

There are way more scenarios than what I have here.
What this provides is a way to test out SASL and SSL authentication, PLAINTEXT and SSL encryption, and basic-auth and no-auth authentication of the RESTful endpoints of Confluent’s Schema Registry and Kafka Connect. Also, a configuration I had yet to try, a custom OAUTH implementation. From a tooling standpoint, I also wanted to also see if a tool can handle multiple connect clusters associated with a given Kafka cluster. This is why one cluster has two separate connect clusters; to verify tools that support cluster information supports having multiple of them (with different Kafka listeners).

Cluster 1: Non-Authenticated Cluster

Connectivity
- Kafka Brokers
- PLAINTEXT protocol on port 9092
- SSL protocol on port 9093
- Schema Registry
- HTTP
- Kafka Connect
- HTTP

Configuration
- 1 Zookeeper
- 3 Brokers
- Inner Broker Protocol PLAINTEXT
- 1 Schema Registry
- Broker Protocol PLAINTEXT
- Kafka Connect Cluster A
- Broker Protocol PLAINTEXT
- Kafka Connect Cluster B
- Broker Protocol SSL

Cluster 2: SASL Authenticated Cluster

Connectivity
- Kafka Brokers
- SASL_PLAINTEXT protocol on port 9092 (scram-sha-512 & plain)
- SASL_SSL protocol on port 9093 (scram-sha-512)
- Schema Registry
- HTTPS with Basic Authentication
- Kafka Connect
- HTTPS with Basic Authentication

Configuration
- 1 Zookeeper
- 3 Brokers
- Inner Broker Protocol SASL_PLAINTEXT (plain)
- Schema Registry
- Broker Protocol SASL_PLAINTEXT (plain)
- Kafka Connect
- Two Workers
- Broker Protocol SASL_PLAINTEXT (plain)

Cluster 3: SSL Authenticated Cluster

Connectivity
- Kafka Brokers
- SSL protocol on port 9093 (SSL Client Authentication required)
- Schema Registry
- HTTPS with Basic Authentication
- Kafka Connect
- HTTPS with Basic Authentication

Configuration
- 1 Zookeeper
- 3 Brokers
- Inner Broker Protocol SSL
- Schema Registry
- Broker Protocol SSL
- Kafka Connect
- Broker Protocol SSL

Cluster 4: OAUTH Authenticated Cluster

Connectivity
- Kafka Brokers
- SASL_PLAINTEXT protocol on port 9092 (plain)
- SASL_SSL protocol on port 9093 (oauthbearer)

Configuration
- 1 Zookeeper
- 3 Brokers
- Inner Broker Protocol SASL_PLAINTEXT (plain)
- No Schema Registry
- No Connect Cluster
- Open Source Hydra OAuth Server
- Image extended to pre-configure database and users created.
- OAuth Client java library

Demonstration Project

The above clusters and configurations are freely available in an Apache 2.0 Licensed project on GitHub. This licensing does not apply to the components being evaluated, and you need to validate and understand their licensing before bringing them into your organization.

Project Notes

If clusters fail to start, the first thing to check is if you created the certificates. If certificates don’t exist the brokers will fail to start and the dependencies on them will cause them to hang.
Because of having multiple clusters at once, port mapping is not done. Use the jumphost container to gain access to kafka command line tools to the clusters.
This project is being developed on a Mac M1 Max with 64GB, with 32GB dedicated to Docker. On a machine with less resources, consider testing against each cluster one at a time.
As I add in new tools, I expect I will have to go back and make some changes. Hopefully not, but with all nuances between them and uncovered copy and paste issues; I expect I will find a few more.

Classifications

Tool classifications, for the sake of these reviews, are monitoring, observation, and administration.

Example operations for each:

Monitoring
- Bytes in and out on a given topic
- Disk utilization of brokers
- Number of partitions on a broker
- Connector status and health
Observation
- Inspect Messages on a topic
- Inspect a Schema
- Consumer Lag
Administration
- Move partitions of a topic
- Change the retention time of a topic
- Adjust offsets of a consumer group
- Pause and resume a connector
- Delete a schema

Tools can support multiple operations, but also know that tools are developed with a specific feature set in mind. I expect this list to change as I learn more about each tool as I attempt to integrate them into these clusters.

Tools

These are current tools planned to be evaluated against these clusters. When the review is completed, a link will be made available from here to that review. The review is more than just an overview of the features and how they work and integrate with Apache Kafka, but a source to use to see each cluster integration together, hoping at least one of these configurations is close to your setup making it quicker for integration and evaluation.

AKQH
CMAK
Grafana
Kafdrop
KafkaUI
kowl
Lensesio

The order listed does not indicate the order of review.

Coming Soon

First I will explore Grafana, which leveraging Prometheus and the Prometheus JMX Exporter. Then I will explore Kafka-UI.

Questions You May have

Why is a tool missing from the list?

This list is currently based on tools that are open-source with free-to-use licensing. Please, feel free to suggest additional tools for evaluation. If the tool is commercial, a free developer license is required for my review. That license should be renewable, as I hope to reevaluate when major changes are released to Apache Kafka (e.g. KRaft) and to the respective tool.

Will you succeed?

I do not know yet the configuration of each tool. I may "give up" saying it is not possible only learn later I just didn’t know how to do it. I will be more than happy to update the process, admit my mistake, and make it easier for others to integrate tools with a given setup.

When will the results of each tool be posted?

My timeline will not be consistent. I have other areas of interest to write about so I expect other articles to be intermixed with this process. Also, I expect there to be a fair amount of work on my part to understand and test out things – well before I even write about it. I can say I have two integrations nearly completed; but I still have to write them.

What about the all the interesting stuff you learned along the way?

The technology questions I have uncovered during this process (like the 3 I mentioned above) will be covered throughout the set of articles.

Reach out

Please contact us if you have ideas, suggestions, want clarification, or just want to talk Apache Kafka administration.

Building an arm64 container for Apache Druid for your Apple Silicon

Neil Buesing — Thu, 08 Sep 2022 13:35:02 +0000

The published Apache Druid container at Docker is a linux/amd64 only image. Running this on your Apple Silicon (M1 or M2 chipset) is slow.

Fortunately, it is super easy to build your own leveraging the binary distribution and existing docker.sh.

All of this is available in a Dockerfile and build script in the druid-m1 repository. The build.sh builds an arm64 image based on the version to be downloaded in the Dockerfile.

A linux/amd64 container-based deployment of Apache Druid on the Apple M1 Silicon takes 2 minutes (1:58.58 - test on Apple M1 Max with 64GB memory and 32GB allocated) to start and become available for processing.
An image built with linux/arm64 based linux images only takes 18 seconds (0:17.79) to become available.

Just need an arm64/v8 image, just download the druid-m1 project and run the build.sh script. What to know a little bit into how it was put together; continue on.

Image

The process of creating this image isn't complicated. Three major pieces went into it's creation.

OS Architecture

First find and use containers that have an arm64/v8 image. Both "openjdk:11-jre-slim" and "busybox" have arm64/v8
images.

Software Installation

The Dockerfile downloads and installs Druid and downloads and uses the druid.sh that is being maintained by Apache Druid.

ARG DRUID_VERSION=0.23.0
ADD https://dlcdn.apache.org/druid/${DRUID_VERSION}/apache-druid-${DRUID_VERSION}-bin.tar.gz /tmp
ADD https://raw.githubusercontent.com/apache/druid/${DRUID_VERSION}/distribution/docker/druid.sh /druid.sh

Druid Extensions

Druid extensions are added by pull-deps operation available with Druid. For this build, the kafka-emitter extension is included, but others are easy to add.

RUN \
    java -cp "/opt/druid/lib/*" \
        -Ddruid.extensions.directory="/opt/druid/extensions/" \
        -Ddruid.extensions.hadoopDependenciesDir="/opt/druid/hadoop-dependencies/" \
        org.apache.druid.cli.Main tools pull-deps --no-default-hadoop \
        -c "org.apache.druid.extensions.contrib:kafka-emitter"

Why The Difference?

The Dockerfile that is part of Apache Druid is all about building the software. But since this is being done after a build is released; its approach is to used the fact that the binaries
are available for download.

New To Druid?

If you are new to Druid and want to see what it can do, check out the druid-late demonstration within dev-local-demos. It leverages a container-based ecosystem provided at dev-local. Update the .env file within the druid folder to point to your individually built arm64 image.

Reach Out

Please contact us if you would like to talk about online analytic processing or event-streaming.

Showcasing Change Data Capture with Debezium and Kafka

Neil Buesing — Wed, 07 Sep 2022 12:57:23 +0000

Setting up Change Data Capture with Databases, Apache Kafka, Kafka Connect, and Debezium takes time; with tricky configurations along the way.

Here we will walk through a setup of all components to showcase what is possible and give the complete picture to give you the pieces to bring this into your project.

Real-Time Ecosystem

This demo is available in the rdbms-cdc-nosql folder in dev-local-demos project. It leverages applications available through containers in dev-local. Within a few minutes, you can see change data captures from relational databases (Postgres, MySQL v8, and MySQL v5) into NoSQL data stores (Mongo, Cassandra, and Elastic).

Stream enrichment processing is done with ksqlDB.

Challenges

There are many nuances in setting up change-data-capture, and doing this within an enterprise organization takes effort and time.

The challenges address here are an attempt to make this a little be easier by providing you a complete POC demonstration to see the steps needed to make a change-data-capture solution.

The specific touch-points here covered:

Enabling logging within a database
Connector setting nuances
Management of Kafka Connect Secrets
Logical Data-types

Database logging

Each database has its own nuances to set up logging. This is critical for any change-data-capture process, and something that needs to be well understood for success. Here are the settings and issues in the configuration of Postgres and MySQL with Debezium for change data capture. This is not a complete overview of all the settings, but rather to provide insight into the complexities that need to be determined between your database operations and development teams. Work with your database administration to enable database logging on the tables that are needed, along with any snapshotting or specific configurations.

Postgres and Debezium

TL;dr;
- wan_level must be set to logical.
- If you are running postgres with docker compose, override command to the following:
```
command: "postgres -c wal_level=logical"
```
- connector needs to enable the pgoutput plugin.
- Add the following to the connector configuration:
```
"plugin.path" : "pgoutput"
```
Details
- Postgres needs to be able to capture changes; this is done through the write-ahead-log (wal). The amount of data captured is based on the wal_level settings. The default setting is replica but this is insufficient level of data for debezium. The logical setting includes replica information and additional logical change sets.
- Debezium has to be configured to use the pgoutput plugin. Use the configuration property plugin.name to set this.
Troubleshooting

This is a set of errors seen when using Debezium Postgres Source Connector.

Attempting to start debezium with Postgres, w/out wal_level properly defined.

Connector configuration is invalid and contains the following 1 error(s):  
Postgres server wal_level property must be \"logical\" but is: replica

Restarting postgres w/out -c wal_level=logical will result in postgres failing to start with the following error:
```
FATAL:  logical replication slot "debezium" exists, but wal_level < logical
```

Starting a connector w/out pgoutput plugin enabled.

io.debezium.DebeziumException: Creation of replication slot failed

If (I should say when) you uncover an error; take the time to document and document it well. Having a deja-vu moment when an error re-surfaces, is not fun.

MySql (v8) and Debezium

TL;dr;
- Version 8 of MySql has logging enabled by default. In production, however, you do need to verify this with operations.
Troubleshooting
- You delete and recreate your MySQL database, but reuse the connector (and the state it persists in Kafka)

Caused by: io.debezium.DebeziumException: 
Client requested master to start replication from position > file size Error code: 1236; SQLSTATE: HY000

MySql (v5) and Debezium

TL;dr;
- Binlog is not enabled by default, but is rather simple to do, set log-bin name and configure binlog_format to row.
- You need to ensure logs are maintained for as longer than any level of reprocessing.
- Debezium expects to resume where it has left off, modifying the logs after debezium has been started can lead to some unexpected errors.
Details
- Add the following properties to your database's mysql.cnf file.

  server-id         = 1
  log_bin           = mysql-bin
  expire_logs_days  = 99
  binlog_format     = row

Troubleshooting

you recreate your database but use the same instance of the connector (v5 has same error as v8).

Caused by: io.debezium.DebeziumException: 
Client requested master to start replication from position > file size Error code: 1236; SQLSTATE: HY000.

MySQL was shut down or became inaccessible. This is a pretty easy fix, just not intuitive.

Caused by: io.debezium.DebeziumException: Failed to read next byte from position XXXXX

Debezium

Debezium is an excellent change-data-capture open-source product. It provides a lot of feature that make it a powerful tool. I find that having a few working examples make it a lot easier to understand. Here is a few things pointed out as understanding this made it easier to configure them and quickly get to seeing the rewards of change data capture.

database.server.name
- This property is not a connection property to the database, but rather the name used to keep this connection uniquely identified. It is used in the topic name generated for the CDC process against the source database. My suggestion is to not pick the type of database as the name (e.g. postgres or mysql). Picking a name like this could cause those maintaining the code to believe this name needs to align to the type of the database.
io.debezium.transforms.ExtractNewRecordState
- By default, Debezium provides nested elements of before, after, and operation. For most use-cases, extracting just the after state is sufficient, and Debezium provides a Single Message Transform (SMT) to do just that. Nested elements can be tricky if you are not writing stream applications; so allowing the data to be flattened with one simple SMT is very helpful. Using this SMT makes it easier to pull data into ksqlDB for enrichment.
Predicates (Apache Kafka 2.6)
- Prior to Apache Kafka 2.6, transformations were unconditional, making a single connector process multiple tables more difficult. By using predicates, a single connector can have different rules for extracting the key from the message.
- A common use of SMTs in Debezium connectors is to pull out the element from the value that is the primary-key, this is to ensure that the events on a given row (primary-key) in the database are processed in order.
database.history
- Many connectors allow for metadata related to the connector to be sourced to a different Kafka cluster. This flexibility leads to confusion, especially to developers new to Kafka Connector and a specific Connector.
- Debezium's database history, is designed this way. You need to set-up bootstrap server, protocol, and any other connection to the kafka cluster to maintain this information, even if it is the same cluster. For enterprise deployments, this flexibility is critical. For proof-of-concepts, development, and trying to get something up and running quickly it is a lot of duplicate configuration.
decimal.handling.mode
- Setting this to string can address sink connector issues that cannot handle the decimal logical type. The demo code uses ksqlDB to cast decimal to string, for downstream sinks that need the help, but this is an alternative approach.

Connector Secrets

Apache Kafka provides a Configuration plugin interface that allows for secrets to be stored separately from configuration. This is also available to a distributed connect cluster, and accessible from the connectors.
config a file connector as follows org.apache.kafka.common.config.provider.FileConfigProvider in the settings of the distributed connect cluster.

config.providers=file
config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider

In addition to leveraging these from your kafka component configurations, they are also accessible from connectors; such as:

"username" : "${file:/etc/kafka-connect/secrets/mysql.properties:USERNAME}"

Put more than secrets in this file; if you store the URL connecting string to the database here, but connection strings and other settings that vary between deployment environments. By keeping those removed from the configuration, a single artifact than be published and maintained with your source-code.

"connection.url" : "${file:/etc/kafka-connect/secrets/mysql.properties:CONNECTION_URL}",
"connection.user" : "${file:/etc/kafka-connect/secrets/mysql.properties:CONNECTION_USER}",
"connection.password" : "${file:/etc/kafka-connect/secrets/mysql.properties:CONNECTION_PASSWORD}"

Schemas and Data-types

When it comes to data-types, especially those consider to be logical data types in Connect API, not all connectors are the same. If you are doing change data capture, odds are you will have decimal's and timestamps. Fortunately, timestamps are stored as long epoch, which will usually translate into a database even if the logical type is not properly handled. Decimals, however, are stored as byte arrays.If the connector doesn't properly invoke the logical converter, it will not be properly converted for the end system. What makes matter worse, connectors are not consistent on how they handle the errors.

Specific Connector Observations

Mongo

In this demonstration, MongoDB Sink Connector properly handles the logical type, and data is stored correctly. The SchemaRecordConverter properly handles the conversion; but as you can see, the converter has to account for and handle logical-types; it is not being done within the Connect API.

Data shown in MongoDB.

Elasticsearch

If elasticsearch sink connector creates the index in elastic (schema.ignore=false), logical types are handled properly. If the sink connector doesn't create the index (schema.ignore=true), logical converters are not processes and logical-type decimals will end up as an array of bytes in elasticsearch.

Index generated by the connector. In each record the amount is a decimal value.

Index is built manually through elastic API (connector does not create it). The decimal bytes are passed as-is to the index; yielding in a non-desired result. Each record is the byte array value (the physical representation of the logical type).

The amounts shown in the Kibana screenshot of HUM=, G2Q=, and FIA= are actually the physical byte arrays converted to strings.

Cassandra

The Datastax Cassandra Sink Connector does not handle the decimal logical-type correctly. What makes matter worse, is the conversion is a warning that shows up in the connect cluster log. No data is written, and the connector keeps running. When checking the log, you can see:

WARN Error decoding/mapping Kafka record ... : Codec not found for requested operation: 
[DECIMAL <-> java.nio.ByteBuffer] (com.datastax.oss.kafka.sink.CassandraSinkTask)

For every sink connector you plan to use:

Be sure to test with decimals, timestamps, & dates; unless that truly isn't in your use case.
Don't simplify your POC. For example, don't let elasticsearch sink connector create your indexes in your POC, if that is not possible in production.
Watch the logs and check for WARN or even INFO.
If you have issues with logical types, you can have Debezium use strings for decimals (decimal.handling.mode=string), or you can leverage a stream processor (e.g. ksqlDB) to cast a decimal to a string.

Takeaways

Enabling database logging can be tricky and each database has unique ways of configuration and enabling.
Having an end-to-end proof of concept showcasing change-data-capture is a great way to get developer buy-in and involvement, but plan appropriate time working with your database operational team to get it enabled in the enterprise.
Leverage Apache Kafka's Config Provider for secrets and environment specific differences, such as the database connection string. The ability to check in a single configuration artifact into revision control w/out specifics to a given environment, is a great benefit.
Validate decimals, dates, and timestamps, as not all connectors handle them correctly.
Check out dev-local project and the demo in rdbms-cdc-nosql in dev-local-demos. The specifics discussed here is based on this demonstration, and scripts are there to get you to start observing change-data-capture with Debezium and Kafka within minutes.

Reach out

Please contact us if you have improvements, want clarification, or just want to talk streaming.

Building a Mix Protocol Apache Kafka Cluster

Neil Buesing — Mon, 01 Jun 2020 21:51:59 +0000

Introduction

Goal: build a multi-protocol Apache Kafka Clusters for SSL Client Authentication for all clients while leveraging PLAINTEXT for inter broker communication.

There are many tutorials and articles on setting up Apache Kafka Clusters with different security options. However, I have not seen a mix-protocol cluster with SSL for encryption and client authentication and PLAINTEXT for broker communication; so I decided to build one.

tl;tr

Setup broker configuration of super.users for SSL authenticated users, not for PLAINTEXT.
Start brokers with SSL inter broker communication.
create full-access ACLs for user User:ANONYMOUS, but only for the IPs of your brokers.
Change brokers to use PLAINTEXT inter broker communication.
Do a rolling restart.

Challenges

The main challenges in setting up the cluster were:

Kafka broker authorizers are not protocol specific.
Unlike SASL, SSL authorization cannot be done over PLAINTEXT; it has to be over SSL encryption.
SSL Certificates are tricky, and those were addressed in What I learned about SSL Certificates when building a Secured Kafka Cluster.
While ACLs can allow resource access to a given principal from a given set of hosts, the broker's super.users is only a principal
setting.

Configuration

SSL

Each broker needs to be configured with the truststore and keystore referencing the appropriate jks files.

    ssl.key.credentials = kafka.key
    ssl.keystore.credentials = kafka.key
    ssl.keystore.filename = broker-{ID}.keystore.jks
    ssl.truststore.credentials = kafka.key
    ssl.truststore.filename = kafka.server.truststore.jks

Note: Confluent platform docker images has minor changes over a standard
deployment; as in the use of ssl.*.credentials and files with those credentials instead of ssl.*.passwords.

Protocols and Listeners

The settings for protocols and listeners are standard. Both SSL and PLAINTEXT are needed.

listener.security.protocol.map

The default value for this broker property is just fine, but if you prefer to only have configured what is needed;
consider setting it to:

```
listener.security.protocol.map = PLAINTEXT:PLAINTEXT,SSL:SSL
```

advertised.listeners

Each broker needs to established its list of listeners and advertises them in a way that the clients and other brokers can
directly access them. Standard 9092 and 9093 ports for PLAINTEXT and SSL used respectively.

  advertised.listeners = PLAINTEXT://broker-{ID}:9092,SSL://broker-{ID}:9093

Authorizer

authorizer.class.name

To enable client authorization, set authorizer.class.name
to kafka.security.authorizer.AclAuthorizer, the provided implementation of org.apache.kafka.server.authorizer.Authorizer.

  authorizer.class.name = kafka.security.authorizer.AclAuthorizer

SSL Client Authentication

ssl.client.auth

Set ssl.client.auth to required or requested.

  ssl.client.auth = required

While you could set this to requested that would still allow SSL clients to not support client authentication. Unless
you have a specific reason to need to do this, make sure you set this to required.

No ACL Found

allow.everyone.if.no.acl.found

Be sure to exclude the property allow.everyone.if.no.acl.found or set to false. If you set this to true,
it allows full access to a resource if no ACLs are found.

  allow.everyone.if.no.acl.found = false

Super Users

super.users

Do not add User:ANONYMOUS to super.users. There is no way to restrict privileges to a given host. Instead, only add the
users of the SSL certificates for your brokers and the user you will use to create ACLs. This ensures that the cluster
is always secure. In addition to each broker's certificate added here; add in the root certificate so kafka-acls can be used to create the proper ACL.

The super.users configuration it should not be used for
authentication for PLAINTEXT. When SSL authorization is enabled, the user presented over PLAINTEXT will always be User:ANONYMOUS.
If you add this user to super.users anyone will be able to access your cluster (unless network restrictions prevent it).

  super.users = User:CN=root;User:CN=broker-1;User:CN=broker-2;User:CN=broker-3;User:CN=broker-4

Inter Broker Protocol

security.inter.broker.protocol

This is a tricky setting. First, start the cluster with SSL.

security.inter.broker.protocol = SSL

Once the ACL for the brokers is created it change it to PLAINTEXT.

security.inter.broker.protocol = PLAINTEXT

The cluster will need to be restarted (following a proper rolling restart) for the change to take effect.

Creating the Secure Cluster

With all these pieces in place, the process is straightforward.

Start Cluster
Create ACLs for a given superuser, User:ANONYMOUS only from IPs of brokers.
Change inter broker protocol on all the brokers from SSL to PLAINTEXT
Perform a rolling restart of the cluster

An alternate approach considered was temporarily adding User:ANONYMOUS to the super.users, and remove it after the ACLs
are configured; to use this approach, ensure the cluster is not accessible until User:ANONYMOUS is part of super.users.

Example Cluster

The Kafka SSL Cluster repository provides an example cluster. It deploys on
a single machine leveraging docker-compose. It showcases this configuration, it is not a production-ready cluster.

See Confluent's quickstart documentation to understand
the differences in some of the configurations leveraging the Confluent Platform docker images.

Use the jumphost container to access the cluster. Scripts to easily create the ACLs and test the cluster are mounted to this
container.

Connect to that cluster using docker exec.

docker exec -it kafka_jumphost bash
cd /opt/bin

Example Code

An Example configuration is available at the GitHub repository Kafka SSL Cluster.
It includes scripts to create certificates, script to swap changing protocol from SSL to PLAINTEXT, a
rolling restart script, and a dashboard. Please see its documentation for additional details.

To run commands to create ACLs, you will need a configuration like the following.

security.protocol=SSL
ssl.key.password=dev_cluster_secret
ssl.keystore.location=/certs/root.keystore.jks
ssl.keystore.password=dev_cluster_secret
ssl.truststore.location=/certs/kafka.server.truststore.jks
ssl.truststore.password=dev_cluster_secret

Super User ACL

Once the cluster is up and running with the security inter broker protocol of SSL, create an ACL based superuser
that can only be used by the hosts of the brokers. With this, the unauthenticated user of User:ANONYMOUS has full operational
access to the cluster, but only from the hosts hosting the brokers.

kafka-acls \
 --bootstrap-server broker-1:9093,broker-2:9093,broker-3:9093,broker-4:9093 \
 --command-config ./config/adminclient-config.conf \
 --add \
 --force \
 --allow-principal User:ANONYMOUS \
 --allow-host 10.5.0.101 \
 --allow-host 10.5.0.102 \
 --allow-host 10.5.0.103 \
 --allow-host 10.5.0.104 \
 --operation All \
 --topic '*' \
 --cluster

Performance Testing

Now that we know it is feasible, is a mixed protocol cluster worth it? While using docker-compose to run a cluster on
one machine is not the best way to property test; it does give insights on where it could lead.

acks=1

With acks=1 theoretically, I would expect no change. However, with less encryption going on between brokers, the CPU
load would go down, so a small improvement is expected. My tests show a 7% improvement.

security.inter.broker.protocol	records/sec	MB/sec	ms avg latency
SSL	29,338.56	27.98	970.06
PLAINTEXT	31,512.17	30.06	926.85

acks=all

With acls=all a larger improvement is expected. This is because the time for the insync replicates to replicate should be faster.
The test shows a 44% improvement; which is greater than what I was expecting. I will need to move this to a realistic
configuration and do additional testing. This result is rather promising and justifies doing more realistic cluster testing.

security.inter.broker.protocol	records/sec	MB/sec	ms avg latency
SSL	12,097.04	11.54	2,557.13
PLAINTEXT	17,418.82	16.62	1,732.00

The Test

Here is the performance script I ran with the ACLs generated for the topic.

performance script

  export BOOTSTRAP_SERVERS=broker-1:9093,broker-2:9093,broker-3:9093,broker-4:9093

  kafka-producer-perf-test \
    --num-records 500000 \
    --record-size 1000 \
    --throughput -1 \
    --producer.config ./config/producer-config.conf \
    --producer-props \
      bootstrap.servers=${BOOTSTRAP_SERVERS} \
      acks=all \
      request.timeout.ms=60000 \
      retries=2147483647 \
    --topic $1

ACLs

  kafka-acls \
   --bootstrap-server broker-1:9093,broker-2:9093,broker-3:9093,broker-4:9093 \
   --command-config ./config/adminclient-config.conf \
   --add \
   --allow-principal User:CN=jumphost \
   --allow-host '*' \
   --operation READ \
   --operation WRITE \
   --topic $1

  kafka-acls \
   --bootstrap-server broker-1:9093,broker-2:9093,broker-3:9093,broker-4:9093 \
   --command-config ./config/adminclient-config.conf \
   --add \
   --allow-principal User:CN=jumphost \
   --allow-host '*' \
   --operation ALL \
   --group '*'

Conclusion

I hope this gives you some insights into Kafka security and possible configurations you could explore for performance
improvements. For me it was the journey to create a cluster configuration I hadn't seen before; giving me the opportunity
to fully understand the security configurations of Apache Kafka.

Additional Information

This article was originally published on my professional blog at buesing.dev.

What I learned about SSL Certificates when building a Secured Kafka Cluster

Neil Buesing — Wed, 08 Apr 2020 11:31:06 +0000

To build a multi-protocol Apache Kafka Clusters to allow for SSL Client Authentication with
PLAINTEXT for inter broker communication, I needed to generate both broker and client SSL certificates.
There were many interesting things I learned in this process and wanted to share them.

An upcoming article will be using these certificates for setting up my Secured Apache Kafka Cluster.

Disclaimer

I build certificates to explore the security available in various systems; in this particular scenario, it is
to explore a mixed protocol Kafka Cluster. Please seek advice from your security teams when creating certificates.

Introduction

If you only need to create certificates as part of an enterprise setup, consider actively supported
projects such as Confluent's Ansible Playbooks.

There are excellent articles on SSL Certificates, including Confluent's documentation
and an excellent tutorial.
The documentation and tutorial focuses on SASL client-side authentication (SASL_SSL protocol)
and I uncovered some unique challenges with client-side SSL authentication.

My experience in certificate generation is based on using OpenSSL. There are other means of generating certificates,
so if another means is used for certificate creation, some of these highlights may not apply to you.

My full scripts are in the repository Kafka SSL Cluster.

Naming

Your certificate subject must start with /, and should contain at minimum a CN element, as in /CN=${hostname}.

For broker certificates, have the CN matches the hostname, and then leverage Subject Alternate Names for
ensuring the certificate can be used with and without domain name successfully.

There are no wildcard options with the standard Apache Kafka Authorizer, kafka.security.authorizer.AclAuthorizer.
So if you a complete subject of common name (CN), organization unit (OU), organization (OU), locality (L), state (S),
and country (C) the full name is what will be used as the User in ACLs.

This: User:CN=broker-1,OU=KAFKA,O=COMPANY,L=CITY,ST=MN,C=US

Not This: User:CN=broker-1

JKS key and keystore passwords should be the same

When creating Java key-stores (.jks files) for your JVMs, make sure the keystore password is the same as the key password.
This is due to limitations of the SunX509 KeyManagerFactory. Java Secure Socket Extension (JSSE) Reference Guide.

All keys in the KeyStore must be protected by the same password

I have found a few other documented cases of this with other systems, the most recent is with BitBucket.

Extensions

Be sure the certificates have the proper x509 extensions.

Basic Contraints extension for CA certificate

basicConstraints=CA:TRUE,pathlen:0

If you setup SSL certificates for client authentication, the CA certificate needs the following x509 extension indicating it is
indeed a CA certificate. Without this extension, it will not authenticate client certificates.

For additional information, see x509v3 config.

Subject Alternate Name Extension for Brokers

subjectAltName=DNS:${i}, DNS:${i}.${DOMAIN}

When it comes to SSL Certificates for encryption, it is important that it can have extensions to handle being used with and without
domain names.

Extended Key Usage Extension (for all)

extendedKeyUsage=serverAuth,clientAuth

The public key a certificate had to be noted that it can be used for server and client authentication. For my cluster,
I configured all certificates so they can be used for both server and client authentication. I added both to all
certificates.

Extensions are not Copied from Request to Certificate

By default, extensions added to a request is not propagated to the certificate. In order for extensions to be copied,
you would need copy_extensions = copy added to your openssl.cfg. I was unsuccessful in getting this to work in
MacOS so I updated my scripts to explicitly add extensions when certificate requests are signed.

It is a good reminder if you create certificate requests for another group to sign, be sure to indicate extensions you
need in your certificates.

Check Status after every call to OpenSSL

If you write scripts, make sure you check return status codes after each command and fail on error. I have spent many hours
troubleshooting script errors only to find out that something failed much earlier in the process.

After each command, add the following. Scripts are finicky, especially with loops. So double-check the error handling
works as well. This is the type of check I do after every openssl and every other command.

[ $? -eq 1 ] && echo "failure" && exit

Intermediate Certificates

Most enterprises do not sign machine or applications certificates with their top-level CA certificate; they use an
intermediate certificate. Using an intermediate certificate requires a certificate chain when creating the trust-store.
Personally, I use intermediate certificates, so it is a reminder on how to properly chain them when creating the pkcs12 file.

The openSSL configuration file

The configuration file of OpenSSL, openssl.cnf can vary greatly between Unix systems and even between Linux distributions.

For every command of openssl that uses openssl.cfg provide a configuration file; otherwise, a default file will be used
and it varies greatly between OS distributions. To avoid confusing on how OpenSSL executes your request, I explicitly
provide the openssl.cfg to use. I keep this file with my scripts. Furthermore, if you are scripting your certificate
process, leverage inline files to provide a custom configuration file to each specific command execution; it is then
self-documenting too.

An example of inlining a file concatenation.

-config <(cat ./openssl.cnf <(printf "\n[ext]\nbasicConstraints=CA:TRUE,pathlen:0"))

Check your OpenSSL documentation

OpenSSL varies between Unix variants and even between Linux distributions; please read the manual pages for your specific
version and release of openssl. Also, there could be a bug or limitation with the instance of openssl you are using.
For example, I was not successful leveraging copy_extensions on MacOS.

Conclusion

What I have documented here were the major pieces I had to figure out when building certificates for client authentication
for my Kafka Cluster. It wouldn't surprise me if there are other things I could uncover, especially limiting extensions
only between the broker certificates and client certificates.

If you want a glimpse to the scripts that I will use in part 2 of this blog where I configure my Secured (Mix-Protocol)
Kafka Cluster, you find them all at Kafka SSL Cluster.