Posts Tagged ‘apachekafka’

« Older Entries
Newer Entries »

From bytes to objects: describing Kafka events

Saturday, October 23rd, 2021

The recording of the talk that Kate Stanley and I gave at Kafka Summit Americas is now available.

Events stored in Kafka are just bytes, this is one of the reasons Kafka is so flexible. But when developing a producer or consumer you want objects, not bytes. Documenting and defining events provides a common way to discuss and agree on an approach to using Kafka. It also informs developers how to consume events without needing access to the developers responsible for producing events.

In our talk, we introduced the most popular formats for documenting events that flow through Kafka, such as AsyncAPI, Avro, CloudEvents, JSON schemas, and Protobuf.

We discussed the differences between the approaches and how to decide on a documentation strategy. Alongside the formats, we also touched on the tooling available for the different approaches. Tools for testing and code generation can make a big difference to your day-to-day developer experience.

The talk was aimed at developers who maybe aren’t already documenting their Kafka events or who wanted to see other approaches.


watch the recording on the Kafka Summit website

(more…)

Tags:, , , ,
Posted in tech | Comments Closed

Stock price events for Kafka

Saturday, August 28th, 2021

I’ve made a Kafka Connect source connector for sending “real-time” (not really) stock price events to a Kafka topic.

Choose your stock (e.g. “IBM“) and run the Connector pointed at your topic, and every minute a JSON event will be produced like:

{"open":137.9,"high":137.9,"low":137.9,"close":137.9,"volume":500,"timestamp":1629421200,"datetime":"2021-08-19 20:00:00"}

(more…)

Tags:,
Posted in code | Comments Closed

Describing Kafka security in AsyncAPI

Tuesday, June 29th, 2021

The new version of AsyncAPI, 2.1.0, was released today. One of the updates is that it lets you describe Kafka security mechanisms and protocols. In this post, I’ll show how you can do this, and how it relates to configuring a Kafka client.

Kafka config means AsyncAPI
sasl.mechanism security.protocol encryption? auth? server protocol security scheme type
unset PLAINTEXT no no kafka
PLAIN SASL_PLAINTEXT no yes, using SASL/PLAIN kafka plain
SCRAM-SHA-256 SASL_PLAINTEXT no yes, using SASL/SCRAM kafka scramSha256
SCRAM-SHA-512 SASL_PLAINTEXT no yes, using SASL/SCRAM kafka scramSha512
OAUTHBEARER SASL_PLAINTEXT no yes, using OAuth kafka oauth2
GSSAPI SASL_PLAINTEXT no yes, using GSSAPI kafka gssapi
unset SSL yes no kafka-secure
PLAIN SASL_SSL yes yes, using SASL/PLAIN kafka-secure plain
SCRAM-SHA-256 SASL_SSL yes yes, using SASL/SCRAM kafka-secure scramSha256
SCRAM-SHA-512 SASL_SSL yes yes, using SASL/SCRAM kafka-secure scramSha512
OAUTHBEARER SASL_SSL yes yes, using OAuth kafka-secure oauth2
GSSAPI SASL_SSL yes yes, using GSSAPI kafka-secure gssapi
unset SSL yes yes, using mutual TLS kafka-secure X509

(more…)

Tags:, ,
Posted in code | Comments Closed

How to avoid SSL handshake errors in your Kafka client because of a self-signed cluster CA

Sunday, June 27th, 2021

You’re trying to connect a Kafka client to a development Apache Kafka cluster which has been quickly set up using a self-signed CA certificate. You don’t have a copy of that CA certificate, and (because it’s not signed by a well-known CA) your Kafka client is failing because of SSL handshake errors.

The error contains messages like
org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
and
javax.net.ssl.SSLHandshakeException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target.

For example:

$ ./bin/kafka-console-consumer.sh \
 --bootstrap-server dale-kafka-saslscram-bootstrap-strimzi.apps.eem-test-fest-6.cp.fyre.ibm.com:443 \
 --topic DALE.TOPIC \
 --group dalegrp \
 --consumer-property 'security.protocol=SASL_SSL' \
 --consumer-property 'sasl.mechanism=SCRAM-SHA-512' \
 --consumer-property 'sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="dale-user" password="pSRtfwTMKNlz";'

[2021-06-27 23:19:06,048] ERROR [Consumer clientId=consumer-dalegrp-1, groupId=dalegrp] Connection to node -1 (dale-kafka-saslscram-bootstrap-strimzi.apps.eem-test-fest-6.cp.fyre.ibm.com/9.46.199.58:443) failed authentication due to: SSL handshake failed (org.apache.kafka.clients.NetworkClient)
[2021-06-27 23:19:06,049] WARN [Consumer clientId=consumer-dalegrp-1, groupId=dalegrp] Bootstrap broker dale-kafka-saslscram-bootstrap-strimzi.apps.eem-test-fest-6.cp.fyre.ibm.com:443 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
[2021-06-27 23:19:06,069] ERROR Error processing message, terminating consumer process:  (kafka.tools.ConsoleConsumer$)
org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
Caused by: javax.net.ssl.SSLHandshakeException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
	at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:131)
	at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:326)
	at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:269)
	at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:264)
	at java.base/sun.security.ssl.CertificateMessage$T13CertificateConsumer.checkServerCerts(CertificateMessage.java:1339)
	at java.base/sun.security.ssl.CertificateMessage$T13CertificateConsumer.onConsumeCertificate(CertificateMessage.java:1214)
	at java.base/sun.security.ssl.CertificateMessage$T13CertificateConsumer.consume(CertificateMessage.java:1157)
	at java.base/sun.security.ssl.SSLHandshake.consume(SSLHandshake.java:392)
	at java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:444)
	at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1074)
	at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1061)
	at java.base/java.security.AccessController.doPrivileged(AccessController.java:770)
	at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask.run(SSLEngineImpl.java:1008)
	at org.apache.kafka.common.network.SslTransportLayer.runDelegatedTasks(SslTransportLayer.java:430)
	at org.apache.kafka.common.network.SslTransportLayer.handshakeUnwrap(SslTransportLayer.java:514)
	at org.apache.kafka.common.network.SslTransportLayer.doHandshake(SslTransportLayer.java:368)
	at org.apache.kafka.common.network.SslTransportLayer.handshake(SslTransportLayer.java:291)
	at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:173)
	at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:543)
	at org.apache.kafka.common.network.Selector.poll(Selector.java:481)
	at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:561)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:265)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:236)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:215)
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:244)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:480)
	at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1257)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1226)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1206)
	at kafka.tools.ConsoleConsumer$ConsumerWrapper.receive(ConsoleConsumer.scala:444)
	at kafka.tools.ConsoleConsumer$.process(ConsoleConsumer.scala:103)
	at kafka.tools.ConsoleConsumer$.run(ConsoleConsumer.scala:77)
	at kafka.tools.ConsoleConsumer$.main(ConsoleConsumer.scala:54)
	at kafka.tools.ConsoleConsumer.main(ConsoleConsumer.scala)
Caused by: sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
	at java.base/sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:439)
	at java.base/sun.security.validator.PKIXValidator.engineValidate(PKIXValidator.java:306)
	at java.base/sun.security.validator.Validator.validate(Validator.java:264)
	at java.base/sun.security.ssl.X509TrustManagerImpl.validate(X509TrustManagerImpl.java:313)
	at java.base/sun.security.ssl.X509TrustManagerImpl.checkTrusted(X509TrustManagerImpl.java:276)
	at java.base/sun.security.ssl.X509TrustManagerImpl.checkServerTrusted(X509TrustManagerImpl.java:141)
	at java.base/sun.security.ssl.CertificateMessage$T13CertificateConsumer.checkServerCerts(CertificateMessage.java:1317)
	... 29 more
Caused by: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
	at java.base/sun.security.provider.certpath.SunCertPathBuilder.build(SunCertPathBuilder.java:141)
	at java.base/sun.security.provider.certpath.SunCertPathBuilder.engineBuild(SunCertPathBuilder.java:126)
	at java.base/java.security.cert.CertPathBuilder.build(CertPathBuilder.java:297)
	at java.base/sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:434)
	... 35 more
Processed a total of 0 messages

I’m assuming that this is just for development purposes, that you know it’s safe to trust the certificate that the Kafka cluster is presenting, and that you’d rather just workaround the error than ask the owner of the Kafka cluster for a copy of their CA.

(more…)

Tags:,
Posted in code | 1 Comment »

Event Endpoint Management

Sunday, June 27th, 2021

Last week, we released the latest version of Event Endpoint Management in IBM Cloud Pak for Integration 2021.2.1. It allows organisations to share and manage access to their Kafka topics. In this post, I want to share a run-through of how it all works.

I’ll start with a high level summary overview, then a walkthrough demo video, and finally share some links to related reading if you’d like more detail.

Overview


click for a larger version of the diagram – numbers in the diagram are described below

Kafka topic owner

This is someone who has a Kafka topic, and is running an application or system that is producing a stream of events to that topic.

They think this stream of events might be useful to other developers in their organisation, so they describe it (using AsyncAPI) and publish this to a catalog where it can be discovered and managed.

  1. creates a Kafka topic and an application that produces events to it
  2. describes and documents their Kafka topic, and the events that are being produced to it
  3. publishes the description of their Kafka topic
  4. pushes the Kafka cluster security info to the Event Gateway service so it can manage access to the topic for the topic owner

App developer

This is someone who is building an application that could benefit from a stream of events.

They are able to discover the event sources that have been shared in their organisation, and get access to them through a self-service Developer Portal.

  1. creates credentials for use in their application
  2. registers new application credentials
  3. updates the Event Gateway service with the new application credentials
  4. creates or configures an application with guidance from the Portal
  5. application connects to the Event Gateway service
  6. application connection routed securely to the Kafka brokers

(more…)

Tags:, ,
Posted in ibm | Comments Closed

A Kafka Developer’s Guide to AsyncAPI

Tuesday, March 30th, 2021

How Kafka developers can use the AsyncAPI specification to describe how their applications are using Kafka topics.

In my post “Why should you document your Kafka topics?” last week, I wrote about the benefits of documenting your Kafka event sources, and mentioned a few of the problems that this can help with.

In this post, I want to show you how you can document the API for your Kafka event sources by creating AsyncAPI documents.

You don’t necessarily have to learn the AsyncAPI specification – tools such as the new Event Endpoint Management capability that I work on in Cloud Pak for Integration make it easy to document APIs with user-friendly forms that generate AsyncAPI documents for you. However, some developers will want to know more about what is happening under the covers, so here is an introduction.


youtu.be/Ni5tCY9r0TY

(more…)

Tags:, ,
Posted in code | Comments Closed

Migrating your Apache Kafka cluster using MirrorMaker 2

Wednesday, March 24th, 2021

You have a Kafka cluster that you have been using for a while. Your cluster has many topics, and the topics have many messages.

Now you’ve decided to move and start using a new, different Kafka cluster somewhere else.

How can you take your topics with you?

Huge thanks to Andrew Borley for co-writing this with me. Useful insights in here probably came from him, the mistakes from me.

(more…)

Tags:, ,
Posted in code | Comments Closed

Describing Kafka with AsyncAPI

Friday, November 27th, 2020

In this post, I want to describe how to use AsyncAPI to document how you’re using Apache Kafka. There are already great AsyncAPI “Getting Started” guides, but it supports a variety of protocols, and I haven’t found an introduction written specifically from the perspective of a Kafka user.

I’ll start with a description of what AsyncAPI is.

“an open source initiative … goal is to make working with Event-Driven Architectures as easy as it is to work with REST APIs … from documentation to code generation, from discovery to event management”

asyncapi.com/docs

The most obvious initial aspect is that it is a way to document how you’re using Kafka topics, but the impact is broader than that: a consistent approach to documentation enables an ecosystem that includes things like automated code generation and discovery.

(more…)

Tags:, , , ,
Posted in code | 3 Comments »

« Older Entries
Newer Entries »