Kafka’s Connect API is a wondrous way of easily bringing data in and out of Apache Kafka without having to write a line of code. By choosing a Connector from the many available, it’s possible to set up and end-to-end data pipeline with just a few lines of configuration. You can configure this by hand, or you can use the Confluent Control Center, for both management and monitoring: BUT … there are times when not all goes well - perhaps your source has gone offline, or one of your targets has been misconfigured.

A very silly PEBCAK problem this one, but Google hits weren’t so helpful so here goes. Running a console producer, specifying keys: kafka-console-producer \ --broker-list localhost:9092 \ --topic test_topic \ --property parse.key=true \ --property key.seperator=, Failed when I entered a key/value: 1,foo kafka.common.KafkaException: No key found on line 1: 1,foo at kafka.tools.ConsoleProducer$LineMessageReader.readMessage(ConsoleProducer.scala:314) at kafka.tools.ConsoleProducer$.main(ConsoleProducer.scala:55) at kafka.tools.ConsoleProducer.main(ConsoleProducer.scala) kafka.common.KafkaException: No key found on line … but I specified the key, didn’t I?

By default, the kafka-avro-console-producer will assume that the schema registry is on port 8081, and happily connect to it. Unfortunately, this can lead to some weird errors if another process happens to be listening on port 8081 already! [oracle@bigdatalite tmp]$ kafka-avro-console-producer \ > --broker-list localhost:9092 --topic kudu_test \ > --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"id","type":"int"},{"name":"random_field", "type": "string"}]}' {"id": 999, "random_field": "foo"} org.apache.kafka.common.errors.SerializationException: Error registering Avro schema: {"type":"record","name":"myrecord","fields":[{"name":"id","type":"int"},{"name":"random_field","type":"string"}]} Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Unexpected character ('<' (code 60)): expected a valid value (number, String, array, object, 'true', 'false' or 'null') at [Source: sun.

I hit this error running Kafka Connect HDFS connector from Confluent Platform v3.1.1 on BigDataLite 4.6: [oracle@bigdatalite ~]$ connect-standalone /etc/schema-registry/connect-avro-standalone.properties /etc/kafka-connect-hdfs/quickstart-hdfs.properties [...] Exception in thread "main" java.lang.IncompatibleClassChangeError: Implementing class at java.lang.ClassLoader.defineClass1(Native Method) at java.lang.ClassLoader.defineClass(ClassLoader.java:763) at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142) at java.net.URLClassLoader.defineClass(URLClassLoader.java:467) at java.net.URLClassLoader.access$100(URLClassLoader.java:73) at java.net.URLClassLoader$1.run(URLClassLoader.java:368) at java.net.URLClassLoader$1.run(URLClassLoader.java:362) at java.security.AccessController.doPrivileged(Native Method) at java.net.URLClassLoader.findClass(URLClassLoader.java:361) at java.lang.ClassLoader.loadClass(ClassLoader.java:424) at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331) at java.lang.ClassLoader.loadClass(ClassLoader.java:357) at java.lang.ClassLoader.defineClass1(Native Method) at java.lang.ClassLoader.defineClass(ClassLoader.java:763) The fix was to unset the CLASSPATH first: unset CLASSPATH

Similar to the previous issue, the sample config in the docs causes another snafu: OGG-15051 Java or JNI exception: oracle.goldengate.util.GGException: Class not found: "kafkahandler". kafkahandler Class not found: "kafkahandler". kafkahandler This time it’s in the kafka.props file: gg.handler.kafkahandler.Type = kafka Should be gg.handler.kafkahandler.type = kafka No capital T in Type! (Image credit: https://unsplash.com/@vanschneider)

In the documentation for the current release of Oracle GoldenGate for Big Data (12.2.0.1.1.011) there’s a helpful sample configuration, which isn’t so helpful … [...] gg.handler.kafkahandler.ProducerRecordClass = com.company.kafka.CustomProducerRecord [...] This value for gg.handler.kafkahandler.ProducerRecordClass will cause a failure when you start the replicat: [...] Class not found: "com.company.kafka.CustomProducerRecord" [...] If you comment this configuration item out, it’ll use the default (oracle.goldengate.handler.kafka.DefaultProducerRecord) and work swimingly! (Image credit: https://unsplash.com/@vanschneider)

There are various reasons for this error, but the one I hit was that the table name is case sensitive, and returned from Oracle by the JDBC driver in uppercase. If you specify the tablename in your connecter config in lowercase, it won’t be matched, and this error is thrown. You can validate this by setting debug logging (edit etc/kafka/connect-log4j.properties to set log4j.rootLogger=DEBUG, stdout), and observe: (I’ve truncated some of the output for legibility)

I’ve been doing some noodling around with Confluent’s Kafka Connect recently, as part of gaining a wider understanding into Kafka. If you’re not familiar with Kafka Connect this page gives a good idea of the thinking behind it. One issue that I hit defeated my Google-fu so I’m recording it here to hopefully help out fellow n00bs. The pipeline that I’d set up looked like this: Eneco’s Twitter Source streaming tweets to a Kafka topic Confluent’s HDFS Sink to stream tweets to HDFS and define Hive table automagically over them It worked great, but only if I didn’t enable the Hive integration part.

Recently added to the oracledi project over at java.net is an adaptor enabling Oracle GoldenGate (OGG) to send data to Elasticsearch. This adds a powerful alternative to [micro-]batch extract via JDBC from Oracle to Elasticsearch, which I wrote about recently over at the Elastic blog. Elasticsearch is a ‘document store’ widely used for both search and analytics. It’s something I’ve written a lot about (here and here for archives), as well as spoken about - preaching the good word, as it were, since the Elastic stack as a whole is very very good at what it does and a pleasure to work with.

I’ve recently been playing around with the ELK stack (now officially known as the Elastic stack) collecting data from an IRC channel with Elastic’s Logstash, storing it in Elasticsearch and analysing it with Kibana. But, this isn’t an “ELK” post - this is a Kafka post! ELK is just some example data manipulation tooling that helps demonstrate the principles. As I wrote about last year, Apache Kafka provides a handy way to build flexible “pipelines”.