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"}
How to use Scratch to create a visualization that explains what parts of an image a TensorFlow image classifier finds the most significant.
An image classifier recognizes this image as an image of The Doctor.
prediction: The Doctor
confidence: 99.97%
Why? What parts of the image did the classifier recognize as indicating that this is the Doctor?
How could we tell?
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 |
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.
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.
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.
A run-through of the DEVELOPMENT.md guide.
In this video, I go from zero to a running Machine Learning for Kids website (including installing all the necessary dependencies and building the site from source).
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”
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.