Oracle to Kafka made easy.
Sometimes we need to expose data from Oracle database to external systems/services.
In our experience, it ends ups with kind of interim tables or even interim databases or REST service solutions, the downsides of such approaches are well known.
These days one of the best solution to communicate services is message broker, namely Apache Kafka.
This time we would like to send data from Oracle database to Kafka, after some googling we come up with – GoldenGate, WebLogic, bunch of connectors and etc… Oh my… all of them proprietary, overcomplicated and heavy, but we just need to capture changes in one table via database trigger, then send them to Kafka.
But it is not (yet?) implemented in native pl/sql, creating REST service on the other hand requires additional “moving parts” and potentially slow.
As we know, communicate to Kafka from Java code is dead simple, so we decided to create our API in Java then load Java code into Oracle stored procedure.
Below you can find the full description of the solution along with source code.
As result we got an a fully functional pl/sql API allowing us easily send data from Oracle database to Apache Kafka without using heavy and complex proprietary software, easier and potentially way faster comparable to custom REST service.
In this article we will show how to publish messages from Oracle Database 12+ to Apache Kafka brokers.
Oracle Database to Apache Kafka integration can be achieved using many methods. In this article we explain how to do this using internal Oracle Aurora JVM. Target audience for this type of integration are sites with active usage of PL/SQL.
I. Installation
I.1 Oracle Database preparation
If you still using Oracle Database 12.1.x you need to configure it to use JDK7. To do that:
perl $ORACLE_HOME/javavm/install/update_javavm_binaries.pl 7
cd $ORACLE_HOME/rdbms/lib
make -f ins_rdbms.mk ioracle
After that for existing instance you need to perform JDK upgrade using $ORACLE_HOME/javavm/install/update_javavm_db.sql script:
$ORACLE_HOME/perl/bin/perl -I $ORACLE_HOME/rdbms/admin $ORACLE_HOME/rdbms/admin/catcon.pl -b jvm_component_upgrade.log $ORACLE_HOME/javavm/install/update_javavm_db.sql
or create new instance using DBCA utility or set of scripts if you are old school Oracle DBA.
You can check version of Oracle Aurora JVM using
select dbms_java.get_jdk_version from dual;
No additional steps are required for Oracle Database 12.2+ including 18c.
I.2 Schema owner setup
Additional Java security permissions required to run Kafka producer code in Oracle Database. We use name APPS (Oracle Applications/Oracle E-Business Suite from 1998):
create user apps identified by apps
default tablespace users
temporary tablespace temp
quota unlimited on users;
grant CONNECT,RESOURCE,JAVAUSERPRIV to APPS;
exec dbms_java.grant_permission( 'APPS', 'SYS:oracle.aurora.security.JServerPermission', 'Verifier', '');
exec dbms_java.grant_permission( 'APPS', 'SYS:java.lang.RuntimePermission', 'getClassLoader', '');
exec dbms_java.grant_permission( 'APPS', 'SYS:javax.management.MBeanTrustPermission', 'register', '');
exec dbms_java.grant_permission( 'APPS', 'SYS:javax.management.MBeanServerPermission', 'createMBeanServer', '');
exec dbms_java.grant_permission( 'APPS', 'SYS:javax.management.MBeanPermission', 'org.apache.kafka.common.metrics.JmxReporter$KafkaMbean#-*', 'registerMBean');
exec dbms_java.grant_permission( 'APPS', 'SYS:javax.management.MBeanPermission', 'org.apache.kafka.common.metrics.JmxReporter$KafkaMbean#-*', 'unregisterMBean');
exec dbms_java.grant_permission( 'APPS', 'SYS:javax.management.MBeanPermission', 'org.apache.kafka.common.utils.AppInfoParser$AppInfo#-*', 'registerMBean');
I.3 Loading of Kafka classes
We use Apache Kafka distribution as source of required libraries.
I.3.A Oracle Database 12.1
Download Apache Kafka 1.1.1 from https://www.apache.org/dyn/closer.cgi?path=/kafka/1.1.1/kafka_2.11-1.1.1.tgz
We recommend this version for Oracle Database 12.1.x, untar downloaded archive and then issue in SQL*Plus as schema owner created on Step I.2
exec dbms_java.loadjava('-r -v -f -noverify /oracle/product/kafka_2.11-1.1.1/libs/lz4-java-1.4.1.jar');
exec dbms_java.loadjava('-r -v -f -noverify /oracle/product/kafka_2.11-1.1.1/libs/log4j-1.2.17.jar');
exec dbms_java.loadjava('-r -v -f -noverify /oracle/product/kafka_2.11-1.1.1/libs/slf4j-api-1.7.25.jar');
exec dbms_java.loadjava('-r -v -f -noverify /oracle/product/kafka_2.11-1.1.1/libs/slf4j-log4j12-1.7.25.jar');
exec dbms_java.loadjava('-r -v -f -noverify /oracle/product/kafka_2.11-1.1.1/libs/kafka-clients-1.1.1.jar');
I.3.B Oracle Database 12.2+
Download Apache Kafka 2.1.0 from https://www.apache.org/dyn/closer.cgi?path=/kafka/2.1.0/kafka_2.12-2.1.0.tgz
(at moment of writing - latest stable Kafka version).
untar downloaded archive and then issue in SQL*Plus as schema owner created on Step I.2
exec dbms_java.loadjava('-r -v -f -noverify /oracle/product/kafka_2.12-2.1.0/libs/lz4-java-1.5.0.jar');
exec dbms_java.loadjava('-r -v -f -noverify /oracle/product/kafka_2.12-2.1.0/libs/log4j-1.2.17.jar');
exec dbms_java.loadjava('-r -v -f -noverify /oracle/product/kafka_2.12-2.1.0/libs/slf4j-api-1.7.25.jar');
exec dbms_java.loadjava('-r -v -f -noverify /oracle/product/kafka_2.12-2.1.0/libs/slf4j-log4j12-1.7.25.jar');
exec dbms_java.loadjava('-r -v -f -noverify /oracle/product/kafka_2.12-2.1.0/libs/jackson-core-2.9.7.jar');
exec dbms_java.loadjava('-r -v -f -noverify /oracle/product/kafka_2.12-2.1.0/libs/jackson-annotations-2.9.7.jar');
exec dbms_java.loadjava('-r -v -f -noverify /oracle/product/kafka_2.12-2.1.0/libs/jackson-databind-2.9.7.jar');
exec dbms_java.loadjava('-r -v -f -noverify /oracle/product/kafka_2.12-2.1.0/libs/kafka-clients-2.1.0.jar');
For any Oracle Database version check for invalid objects:
select count(*) from user_objects where status<>'VALID';
If they exist recompile it using:
exec utl_recomp.recomp_serial;
or
exec utl_recomp.recomp_parallel(<NUMBER_OF_THREADS>);
I.4 Kafka wrapper installation
Download or clone https://github.com/averemee-si/aurora-kafka
Start SQL*Plus and run as schema owner created on Step I.2 downloaded scripts in following order:
@KafkaUtils.sql
@KafkaUtilsJava.pkg
@KafkaUtils.pkg
This will create object type A2_TUPLE which simulates key-value pair, table type A2_ARRAY_OF_TUPLES - array of A2_TUPLE, Java source named eu.solutions.a2.aurora.kafka.KafkaUtils and PL/SQL package A2_KAFKA_UTILS.
Do not forget to check for invalid objects:
select count(*) from user_objects where status<>'VALID';
If they exist recompile using:
exec utl_recomp.recomp_serial;
or
exec utl_recomp.recomp_parallel(<NUMBER_OF_THREADS>);
II. Usage example
Typical usage is in creation of producer with call to A2_KAFKA_UTILS.CREATE_PRODUCER then sending messages with A2_KAFKA_UTILS.SEND_JSON_MESSAGE or A2_KAFKA_UTILS.SEND_STRING_MESSAGE and then closing producer with A2_KAFKA_UTILS.CLOSE_PRODUCER.
declare
OP_RESULT varchar2(2000);
KAFKA_PROPS A2_ARRAY_OF_TUPLES := A2_ARRAY_OF_TUPLES();
MSG_2_SEND A2_ARRAY_OF_TUPLES := A2_ARRAY_OF_TUPLES();
begin
-- Set mandatory Kafka producer properties
KAFKA_PROPS.extend(4);
KAFKA_PROPS(1) := A2_TUPLE('bootstrap.servers', 'kafka1.a2-solutions.eu:9092');
KAFKA_PROPS(2) := A2_TUPLE('client.id', 'KafkaOracleAuroraProducer');
KAFKA_PROPS(3) := A2_TUPLE('key.serializer', 'org.apache.kafka.common.serialization.StringSerializer');
KAFKA_PROPS(4) := A2_TUPLE('value.serializer', 'org.apache.kafka.common.serialization.StringSerializer');
-- Create producer with name TestProducer
OP_RESULT := A2_KAFKA_UTILS.CREATE_PRODUCER('TestProducer', KAFKA_PROPS);
if ('SUCCESS' = OP_RESULT) then
begin
MSG_2_SEND.extend(3);
MSG_2_SEND(1) := A2_TUPLE('INVOICE_ID', '13474');
MSG_2_SEND(2) := A2_TUPLE('INVOICE_NUM', 'AP-2019-13474');
MSG_2_SEND(3) := A2_TUPLE('INVOICE_AMOUNT', '2000');
OP_RESULT := A2_KAFKA_UTILS.SEND_JSON_MESSAGE('TestProducer','TestTopic','Test-Message-Key',MSG_2_SEND);
if ('SUCCESS' = OP_RESULT) then
OP_RESULT := A2_KAFKA_UTILS.CLOSE_PRODUCER('TestProducer');
else
raise_application_error(-20000, OP_RESULT);
end if;
end;
else
raise_application_error(-20000, OP_RESULT);
end if;
end;
/
III. A2_KAFKA_UTILS functions
III.1 CREATE_PRODUCER
Creates Kafka producer
Parameters
PRODUCER_NAME - name of producer
PRODUCER_PROPS - array of tuples with Kafka producer properties
Returns
'SUCCESS' - for successful operation or errorstack in case of error
III.2 CLOSE_PRODUCER
Closes Kafka producer
Parameters
PRODUCER_NAME - name of producer
Returns
'SUCCESS' - for successful operation or errorstack in case of error
III.3 SEND_TEXT_MESSAGE
Sends text message to Kafka broker
Parameters
PRODUCER_NAME - name of producer
TOPIC - Kafka broker topic
MSG_KEY - Message key
MSG_VALUE - Message value
Returns
'SUCCESS' - for successful operation or errorstack in case of error
III.4 SEND_JSON_MESSAGE
JSONifies array of tiples and send this string to Kafka broker
Parameters
PRODUCER_NAME - name of producer
TOPIC - Kafka broker topic
MSG_KEY - Message key
ARRAY_OF_VALUES - Message value
Returns
'SUCCESS' - for successful operation or errorstack in case of error
III.5 ARRAY_2_JSON
Converts array of tuples (A2_ARRAY_OF_TUPLES) to JSON formatted string.
Parameters
ARRAY_OF_VALUES - in, A2_ARRAY_OF_TUPLES
Returns
JSON formatted string presentation of given array of tuples
