This is a short summary discussing what the options are for integrating Oracle RDBMS into Kafka, as of December 2018 (refreshed June 2020). For a more detailed background to why and how at a broader level for all databases (not just Oracle) see this blog and this talk.

What techniques & tools are there? ๐Ÿ”—

Franck Pachot has written up an excellent analysis of the options available here.

As of June 2020, this is what the line-up looks like:

What do they look like in action? ๐Ÿ”—

I did a talk at UK Oracle User Group TECH18 conference, presenting my talk “No More Silos: Integrating Databases and Apache Kafka”. As part of this I did a live demo showing the difference between using the JDBC Connector (query-based CDC) and the Debezium/XStream option (log-based CDC). Here I’ll try and replicate the discussion and examples. You can also see previous articles that I’ve written showing GoldenGate in action.

You can find all of the code on the demo-scene repository, runnable through Docker and Docker Compose. Simply clone the repo, and then run

cd no-more-silos-oracle
./setup.sh

The setup script does all of the rest, including bringing up Confluent Platform, and configuring the connectors. You do have to build the Oracle database docker image first.

Setup ๐Ÿ”—

Some notes on setup of each option:

The actual config of the two connectors is done in separate calls to Kafka Connect’s REST API (JDBC / Debezium).

You can validate that each connector is running by querying the REST API:

curl -s "http://localhost:8083/connectors?expand=info&expand=status" | \
      jq '. | to_entries[] | [ .value.info.type, .key, .value.status.connector.state,.value.status.tasks[].state,.value.info.config."connector.class"]|join(":|:")' | \
      column -s : -t| sed 's/\"//g'| sort

source  |  ora-source-debezium-xstream  |  RUNNING  |  RUNNING  |  io.debezium.connector.oracle.OracleConnector
source  |  ora-source-jdbc              |  RUNNING  |  RUNNING  |  io.confluent.connect.jdbc.JdbcSourceConnector

Initial data load ๐Ÿ”—

In Oracle, check the source data:

COL FIRST_NAME FOR A15
COL LAST_NAME FOR A15
COL ID FOR 999
COL CLUB_STATUS FOR A12
COL UPDATE_TS FOR A29
SET LINESIZE 200
SELECT ID, FIRST_NAME, LAST_NAME, CLUB_STATUS, UPDATE_TS FROM CUSTOMERS;

  ID FIRST_NAME      LAST_NAME       CLUB_STATUS  UPDATE_TS
---- --------------- --------------- ------------ -----------------------------
  1 Rica            Blaisdell       bronze       11-DEC-18 05.16.00.000000 PM
  2 Ruthie          Brockherst      platinum     11-DEC-18 05.16.00.000000 PM
  3 Mariejeanne     Cocci           bronze       11-DEC-18 05.16.00.000000 PM
  4 Hashim          Rumke           platinum     11-DEC-18 05.16.00.000000 PM
  5 Hansiain        Coda            platinum     11-DEC-18 05.16.00.000000 PM

Now let’s see what’s in Kafka. I’m using ksqlDB here to inspect the data; you could use other Kafka console consumers if you’d rather.

Launch ksqlDB:

docker exec -it ksqldb ksql http://ksqldb:8088

Inspect the topics on the Kafka cluster:

ksql> LIST TOPICS;

Kafka Topic               | Registered | Partitions | Partition Replicas | Consumers | ConsumerGroups
-------------------------------------------------------------------------------------------------------
asgard.DEBEZIUM.CUSTOMERS | false      | 1          | 1                  | 0         | 0
ora-CUSTOMERS-jdbc        | false      | 1          | 1                  | 0         | 0
โ€ฆ

The two topics listed are for the same table (CUSTOMERS) from the Debezium and JDBC connectors respectively.

Dump the contents:

Each has the full contents of the source table (5 records, only first is shown above). We can actually use ksqlDB to easily query the topic directly if we want. First we declare each topic as the source for a stream:

SET 'auto.offset.reset' = 'earliest';
CREATE STREAM CUSTOMERS_STREAM_DBZ_SRC WITH (KAFKA_TOPIC='asgard.DEBEZIUM.CUSTOMERS', VALUE_FORMAT='AVRO');
CREATE STREAM CUSTOMERS_STREAM_JDBC_SRC WITH (KAFKA_TOPIC='ora-CUSTOMERS-jdbc', VALUE_FORMAT='AVRO');

and then query the JDBC-sourced Kafka topic:

ksql> SELECT ID, FIRST_NAME, LAST_NAME, CLUB_STATUS FROM CUSTOMERS_STREAM_JDBC_SRC EMIT CHANGES LIMIT 5;

+---------------------+---------------------+---------------------+---------------------+
|ID                   |FIRST_NAME           |LAST_NAME            |CLUB_STATUS          |
+---------------------+---------------------+---------------------+---------------------+
|1                    |Rica                 |Blaisdell            |bronze               |
|2                    |Ruthie               |Brockherst           |platinum             |
|5                    |Hansiain             |Coda                 |platinum             |
|4                    |Hashim               |Rumke                |platinum             |
|3                    |Mariejeanne          |Cocci                |bronze               |

and the one from Debezium:

ksql> SELECT AFTER->ID AS ID, AFTER->FIRST_NAME AS FIRST_NAME, AFTER->LAST_NAME AS LAST_NAME, AFTER->CLUB_STATUS AS CLUB_STATUS FROM CUSTOMERS_STREAM_DBZ_SRC EMIT CHANGES;

+---------------------+---------------------+---------------------+---------------------+
|ID                   |FIRST_NAME           |LAST_NAME            |CLUB_STATUS          |
+---------------------+---------------------+---------------------+---------------------+
|1                    |Rica                 |Blaisdell            |bronze               |
|2                    |Ruthie               |Brockherst           |platinum             |
|3                    |Mariejeanne          |Cocci                |bronze               |
|4                    |Hashim               |Rumke                |platinum             |
|5                    |Hansiain             |Coda                 |platinum             |

Note that I’m accessing nested attributes of the AFTER object here using the -> operator.

The schema for both topics come from the Schema Registry, in which Kafka Connect automatically stores the schema for the data coming from Oracle and serialises the data into Avro (or Protobuf, or JSON Schema). The great thing about this is in a consuming application, such as ksqlDB, the schema is already available and doesn’t have to be manually entered.

INSERT ๐Ÿ”—

Insert a row in the Oracle database:

SQL> SET AUTOCOMMIT ON;
SQL>
SQL> INSERT INTO CUSTOMERS (FIRST_NAME,LAST_NAME,CLUB_STATUS) VALUES ('Rick','Astley','Bronze');

1 row created.

Commit complete.

Straight away in the Kafka topics you’ll see a new row (in fact, if you have left the above SELECT running you won’t need to rerun this, it’ll show the new row already):

So far, so same. Each captures an insert. Debezium from XStream and the database’s redo log, JDBC by polling the database for any rows with a newer UPDATE_TS or higher ID than the previous request.

UPDATE ๐Ÿ”—

This is where things get interesting. Let’s update the row in Oracle that we just created:

SQL> UPDATE CUSTOMERS SET CLUB_STATUS = 'Platinum' where ID=42;

1 row updated.

Commit complete.
SQL>

Now check out the data in Kafka.

DELETE ๐Ÿ”—

Delete a record from the source system

SQL> DELETE FROM CUSTOMERS WHERE ID=42;

1 row deleted.

Commit complete.

Now check out the data in Kafka.

JDBC is unchanged; it’s not captured any change to the source table. If you think about it, this is perfectly reasonable. How you query a database for a row that doesn’t exist? Debezium/XStream, on the other hand, reports the data change precisely:

ksql> SELECT OP, AFTER->ID, BEFORE->CLUB_STATUS AS STATUS_BEFORE, AFTER->CLUB_STATUS AS STATUS_AFTER FROM CUSTOMERS_STREAM_DBZ_SRC EMIT CHANGES;
+---------------------+---------------------+---------------------+---------------------+
|OP                   |ID                   |STATUS_BEFORE        |STATUS_AFTER         |
+---------------------+---------------------+---------------------+---------------------+
|r                    |1                    |null                 |bronze               |
|r                    |2                    |null                 |platinum             |
|r                    |3                    |null                 |bronze               |
|r                    |4                    |null                 |platinum             |
|r                    |5                    |null                 |platinum             |
|c                    |42                   |null                 |Bronze               |
|u                    |42                   |Bronze               |Platinum             |
|d                    |null                 |Platinum             |null                 |

Note the d record on the last row. This has captured the DELETE operation perfectly. The null in the right-most column is the current value for AFTER->CLUB_STATUS, and since the record is deleted, it has no value. We can see this even more clearly if we look at the raw payload for the whole record:

{
  "before": {
    "ID": 42,
    "FIRST_NAME": "Rick",
    "LAST_NAME": "Astley",
    "EMAIL": null,
    "GENDER": null,
    "CLUB_STATUS": "Platinum",
    "COMMENTS": null,
    "CREATE_TS": 1544562543660463,
    "UPDATE_TS": 1544562791000000
  },
  "after": null,
  "source": {
    "version": "1.2.0-SNAPSHOT",
    "connector": "oracle",
    "name": "asgard",
    "ts_ms": 1544563479000,
    "txId": "9.32.712",
    "scn": 3042804,
    "snapshot": true
  },
  "op": "d",
  "ts_ms": 1544563482682,
  "messagetopic": "asgard.DEBEZIUM.CUSTOMERS",
  "messagesource": "Debezium CDC from Oracle on asgard"
}

The full record that has been deleted is present in the BEFORE object, but AFTER is nullโ€”it’s been deleted, it no longer exists. It is an ex-record.

Bonus ksqlDB :

๐Ÿ‘‰ Introduction to ksqlDB

We’re working with data in a Kafka topic. As it happens, ksqlDB is kinda useful for interogating that data, but at the end of the day it’s still just a Kafka topic. We can use ksqlDB to also help monitor the lag between the event in the source system (source->ms_ms as provided by Debezium) and the time recorded on the Kafka broker (the Kafka message timestamp, exposed in ROWTIME):

ksql> SELECT TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss Z'), 
             OP, 
             ROWTIME - SOURCE->TS_MS AS LAG_MS 
        FROM CUSTOMERS_STREAM_DBZ_SRC EMIT CHANGES;
+----------------------------+----------------------------+----------------------------+
|KSQL_COL_0                  |OP                          |LAG_MS                      |
+----------------------------+----------------------------+----------------------------+
|2020-06-12 11:28:50 +0000   |r                           |584                         |
|2020-06-12 11:28:50 +0000   |r                           |570                         |
|2020-06-12 11:28:50 +0000   |r                           |570                         |
|2020-06-12 11:28:50 +0000   |r                           |571                         |
|2020-06-12 11:28:50 +0000   |r                           |570                         |
|2020-06-12 12:34:53 +0000   |c                           |4720                        |
|2020-06-12 12:35:18 +0000   |u                           |18686                       |
|2020-06-12 13:11:36 +0000   |d                           |62802                       |

Some of these lag times are pretty high; DBZ-1018 Oracle connector is laggy is a JIRA currently tracking it.

You can get the same data out of the JDBC connector, based on the UPDATE_TS of the record itself:

ksql> SELECT TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss Z'), 
             ROWTIME - UPDATE_TS AS LAG_MS 
        FROM CUSTOMERS_STREAM_JDBC_SRC  EMIT CHANGES;
+--------------------------------------------+--------------------------------------------+
|KSQL_COL_0                                  |LAG_MS                                      |
+--------------------------------------------+--------------------------------------------+
|2020-06-12 11:20:01 +0000                   |340770                                      |
|2020-06-12 11:20:01 +0000                   |340774                                      |
|2020-06-12 11:20:01 +0000                   |340775                                      |
|2020-06-12 11:20:01 +0000                   |340776                                      |
|2020-06-12 11:20:01 +0000                   |340777                                      |
|2020-06-12 12:34:51 +0000                   |3195                                        |
|2020-06-12 13:08:01 +0000                   |1046                                        |

You’ll note here no available OP information, and no row for the corresponding DELETE action in the source database.

Ecosystem ๐Ÿ”—

When you’re bringing data into Kafka, you need to remember the bigger picture. Dumping it into a topic alone is not enough. Well, it is, but your wider community of developers won’t thank you.

You want to ensure that the schema of the source data is preserved, and that you’re using a serialisation method for the data that is suitable. Doing this means that developers can use the data without being tightly coupled to the producer of the data to understand how to use it.

However you do this, it should be in a way that integrates with the broader Kafka and Confluent Platform ecosystem. One option is the Schema Registry and Avro (or Protobuf, or JSON Schema). If you’re using Kafka Connect then this is available by default, since you just select the Avro converter when you set up Kafka Connect.

Overview of the Pros and Cons of each technique ๐Ÿ”—

Some of these are objective, others subjective. Others may indeed be plain false ;-) Discussion, comments, and corrections via Twitter or LinkedIn welcomed!

References ๐Ÿ”—

Try it out! ๐Ÿ”—

You can find all of the code used in this article on github here.

Feedback? ๐Ÿ”—

Some of these are objective, others subjective. Others may indeed be plain false ;-) Discussion, comments, and corrections in the comment function below welcomed!

For help in getting this working, the best place to head is the Confluent Community:

There’s also a good Debezium community:

Updates & Comments ๐Ÿ”—