(This is an expanded version of the intro to an article I posted over on the Confluent blog. Here I get to be as verbose as I like ;))

My first job from university was building a datawarehouse for a retailer in the UK. Back then, it was writing COBOL jobs to load tables in DB2. We waited for all the shops to close and do their end of day system processing, and send their data back to the central mainframe. From there it was checked and loaded, and then reports generated on it. This was nearly twenty years ago as my greying beard will attest—and not a lot has changed in the large majority of reporting and analytics systems since then. COBOL is maybe less common, but what has remained constant is the batch-driven nature of processing. Sometimes batches are run more frequently, and get given fancy names like intra-day ETL or even micro-batching. But batch processing it is, and as such latency is built into our reporting by design. When we opt for batch processing we voluntarily inject delays into the availability of data to our end users. Much better is to build our systems around a streaming platform instead.

This is the detailed step-by-step if you want to recreate the process I describe in the Confluent blog here

---

I used Oracle’s Oracle Developer Days VM, which comes preinstalled with Oracle 12cR2. You can see the notes on how to do this here. These notes take you through installing and configuring:

• Swingbench, to create a sample “Order Entry” schema and simulate events on the Oracle database
• Oracle GoldenGate (OGG, forthwith) and Oracle GoldenGate for Big Data (OGG-BD, forthwith)
  • I’m using Oracle GoldenGate 12.3.1 which includes the Kafka Connect handler as part of its distribution. A connector for earlier versions can be found here. Some of the syntax may differ in the configuration below - if you hit problems then check out an article that I wrote with an earlier version of the tool.
• OGG extract from the Order Entry schema
• Confluent Platform
• KSQL
• Elasticsearch

From this point, I’ll now walk through configuring OGG-BD with the Kafka Connect handler

---

See also Kafka Listeners - Explained

---

A short post to help Googlers. On a single-node sandbox Apache Kafka / Confluent Platform installation, I was getting this error from Schema Registry, Connect, etc:

WARN [AdminClient clientId=adminclient-3] Connection to node -1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

KSQL was throwing a similar error:

KSQL cannot initialize AdminCLient.

I had correctly set the machine’s hostname in my Kafka server.properties:

Some notes that I made on installing and configuring Oracle GoldenGate with Confluent Platform. Excuse the brevity, but hopefully useful to share!

---

I used the Oracle Developer Days VM for this - it’s preinstalled with Oracle 12cR2. Big Data Lite is nice but currently has an older version of GoldenGate.

Login to the VM (oracle/oracle) and then install some useful things:

sudo rpm -Uvh https://dl.fedoraproject.org/pub/epel/epel-release-latest-7.noarch.rpm
sudo yum install -y screen htop collectl rlwrap p7zip unzip sysstat perf iotop
sudo su -
cd /etc/yum.repos.d/
wget http://download.opensuse.org/repositories/shells:fish:release:2/CentOS_7/shells:fish:release:2.repo
yum install fish

Check Oracle version etc:

There’s an impressive 19 sessions that cover Apache Kafka™ at Oracle OpenWorld, JavaOne, and Oak Table World this year! You can find the full list with speakers in the session catalogs for OOW, JavaOne, and Oak Table World. OTW is an awesome techie conference which is at the same time as OpenWorld, next door to Moscone. Hope to see you there!

Check out the writeup of my previous visit to OOW including useful tips here.

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. What then? Well of course, it’s diagnostics time! And for diagnostics, you need logs. When you launch Kafka Connect it logs everything to stdout, and this output includes content from the Kafka Connect REST interface. This REST interface is for configuration and control of the connectors (status/pause/resume) - and whilst Control Center is being used on the Connect configuration screens, you’ll notice that the REST interface gets polled frequently - every couple of seconds, with a greater number of requests the more connectors you have. All of this goes into the log:

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?

It would help if I could spell … key.seperator isn’t a valid property to configure. separator on the other hand, is:

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.net.www.protocol.http.HttpURLConnection$HttpInputStream@4e0ae11f; line: 1, column: 2]; error code: 50005
        at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:170)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:187)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:238)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:230)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:225)
        at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.registerAndGetId(CachedSchemaRegistryClient.java:59)
        at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:91)
        at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:72)
        at io.confluent.kafka.formatter.AvroMessageReader.readMessage(AvroMessageReader.java:158)
        at kafka.tools.ConsoleProducer$.main(ConsoleProducer.scala:55)
        at kafka.tools.ConsoleProducer.main(ConsoleProducer.scala)

Solution? Make sure you specify the schema URL when you launch the producer, using --property schema.registry.url=http://localhost:18081 :

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. For me the integration with Hive to automatically define schemas was one of the key interests for this platform, so I wanted to see if I could get it to work. The error I got was

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. So, being able to combine that with my “day job” focus of Oracle is fun. Let’s get started!

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”. Today I’m writing up a short real-world example of this in practice. There are three elements to the flexibility that I really want to highlight:

The Oracle by Example (ObE) here demonstrating how to use Goldengate to replicate transactions big data targets such as HDFS is written for the BigDataLite 4.2.1, and for me didn’t work on the current latest version, 4.4.0.

The OBE (and similar Hands On Lab PDF) assume the presence of pmov.prm and pmov.properties in /u01/ogg/dirprm/. On BDL 4.4 there’s only the extract to from Oracle configuration, emov.

Fortunately it’s still possible to run this setup out of the box in BDL 4.4, with bells on because it includes Kafka too. And, who doesn’t like a bit of Kafka nowadays?