What it does

Oracle SQL Access to Kafka Version 2 enables Kafka streaming data to be processed with Oracle Database tables using standard Oracle Database SQL semantics, and enables data to be processed by standard Oracle application logic (for example, Oracle JDBC applications). Oracle SQL access to Kafka is integrated in Oracle Database. This integration of Kafka access in Oracle Database enables you to relate tables in Oracle Database using data streams produced by Kafka or an OCI Streaming Service, without requiring an external client connector application. Oracle SQL access to Kafka can scale up data streams for Oracle Database in the same fashion as Kafka applications.

Oracle SQL access to Kafka enables you to do the following:

• Create and use a streaming application to process unread Kafka records one time, where these records do not need to be retained after they are processed.
• Create and use a loading application to capture unread Kafka records permanently in an Oracle Database table, for access by various Oracle applications. In this case, Kafka records are captured and persisted in user tables in Oracle Database. This use case is helpful for data warehouses.
• Create and use a seeking application to reread records that are in a Kafka topic, based on a user-supplied timestamp interval.
• Create and use two or more streaming applications. These applications can be used to stream data from two or more Kafka topics, where you can then join them using SQL in Oracle Database.

How It Works

Oracle SQL access to Kafka version 2 provides access to Kafka data using Oracle Database system-generated views, and external tables. These views and external tables use the DBMS_KAFKA package to define a named Oracle SQL access to Kafka application. In general, these views and external tables are transparent for streaming, loading, and seeking applications.

Your application can perform and control operations as an Oracle Database transaction, complying with the ACID (Atomicity, Consistency, Isolation, Durability) requirements for the database, ensuring that either all parts of the transaction are committed, or all rolled back, with a unique identifier (a transaction ID) for each transaction. This transaction ID includes timestamps that you can use to identify and roll back errors. The ACID feature of Oracle Database transactions provides support for data recovery in case of a failure, without losing or repeating records.

The Oracle transaction performed with Oracle SQL access to Kafka includes managing the Kafka partition offsets, and committing them to database metadata tables in Oracle Database.

Without Oracle SQL Access to Kafka, the Kafka partition offsets need to be managed either by the application, or by Kafka, neither of which support transaction semantics. This means that after a system failure, Kafka records can be lost or reprocessed by an application. Managing offsets in an Oracle Database transaction avoids these problems, and enhances the isolation and durability of the Kafka data.

Because Oracle SQL Access to Kafka is available with Oracle Database, and is used with PL/SQL and SQL queries, no external client application is required to provide a connector to Oracle Database.

The ORA_KAFKA PL/SQL package has functions and procedures to register a Kafka cluster in a database schema, query Kafka topics, query data from specified offsets or specified timestamps, and more. You can choose either to use global temporary tables without storing the data, or store the data into user tables in the target Oracle Database.

How you can use it

You can use Oracle SQL access to Kafka application to access global temporary tables or user tables created in Oracle Database, so that your application can obtain data. That data can be streams of data, or snapshots of the data from other databases, which can be accessed directly, or loaded into Oracle Database tables and be used within your application.

Kafka global temporary tables have the following characteristics:

• The global temporary table is loaded once at the outset of an application instance, and used as a snapshot of Kafka records for the duration of the application instance. The application can use standard Oracle SQL with the global temporary table.
• Each query from a global temporary table results in a trip to the Kafka cluster, re-retrieving the same rows, and perhaps additional rows.

The corresponding global temporary table receives a snapshot from an Oracle SQL access to Kafka view. Applications use this temporary table for one or more queries within a transaction: a global temp table is loaded once, and used. The Kafka offsets are advanced, and then the app commits, indicating that it is finishee\d with the Kafka records loaded in the global temporary table.

Reading from the temporary table is beneficial for the following reasons:

• Repeatable reads are supported, either explicitly from multiple queries or implicitly within a join
• Reliable statistics are gathered for the query optimizer
• Only one trip is made to Kafka when loading the temporary table. Subsequent queries do not result in a trip to the Kafka cluster.
• Global temporary tables can be joined with standard Oracle tables. Joining Oracle SQL access to Kafka temporary tables with Oracle Database tables increases your ability to use Oracle Database capabilities with Kafka data.
• You can leverage the mature optimization and processing strategies in Oracle Database to minimize code paths needed to join tables efficiently.

The global temporary tables can be joined with standard Oracle tables (for example, Oracle customer relationship management (CRM) tables.

By joining Oracle SQL access to Kafka temporary tables with Oracle Database tables, you obtain the following advantages:

• Leveraging the mature optimization and execution strategies in Oracle Database to minimize code path required to join tables efficiently
• Obtaining Oracle Database transaction semantics, with the security of Oracle Database ACID transaction processing (atomicity, consistency, isolation, and durability), ensuring that all changes to data are performed as if they are a single operation, controlled by the application
• Managing the Kafka partition offsets and committing them to database metadata tables in Oracle Database, so that after a system failure, these Oracle Database transactions with Kafka records are not subject to being lost or reprocessed by an application.

To provide role-based authentication to grant the Oracle SQL access for Kafka administration privileges to an administrative user, Oracle provides the OSAK_ADMIN_ROLE starting with Oracle Database 23ai. You can grant this role to an administrator user for Oracle SQL Access to Kafka. This role grants the system privileges required for users that you designate as Oracle SQL access for Kafka administrators to configure, register, and manage Kafka clusters. The system privileges granted by this role are as follows:

• CREATE CREDENTIAL, to create a Kafka SASL-SSL (Simple Authentication and Security Layer) password or OSS (Oracle Streaming Service) authToken
• CREATE ANY DIRECTORY, to create cluster access and cluster configuration directory
• DROP ANY DIRECTORY, to drop cluster access and cluster configuration directory
• READ privileges to sys.dbms_kafka_clusters
• READ privileges to sys.dbms_kafka_applications
• READ privileges to sys.dbms_kafka_messages

As a DBA, you create and grant users privileges to administer and use Oracle SQL access to Kafka. There are two categories of users:

• Oracle SQL Access to Kafka administrators are privileged users. To simplify management of Oracle SQL access to Kafka, Oracle recommends that the Oracle DBA grant the OSAK_ADMIN_ROLE to designated Kafka administrators. This role is precreated in the database starting with Oracle Database 23ai.

  Administrators run the DBMS_KAFKA_ADM package methods to configure and manage the Kafka cluster information. Either users granted OSAK_ADMIN_ROLE or the Oracle DBA can create the operating system level cluster configuration directory, and populate that directory with configuration files. Oracle SQL Access to Kafka administrators create the Oracle directory object for the Kafka cluster configuration and access directories.

• Application users of Kafka topic data are granted the READ privileges required to access to the DBMS_KAFKA packages, so that they can access and use data accessed from Kafka cluster topics.

Oracle SQL access to Kafka provides the following application modes that you can use to attach to the Apache Kafka cluster:

• Loading: Use to load data from a Kafka Topic into an Oracle Database table.
• Streaming: Use to read sequentially through a Kafka topic.
• Seekable: Use to access a Kafka topic randomly between starting and ending timestamps that you designate.

Secure Kafka Clusters

To maintain securely encrypted data transmission between Oracle Database and clusters, Oracle SQL access to Kafka employs several security protocols. For access to secure Kafka clusters and Oracle Streaming Services (OSS) clusters, security configuration files are used. These operating system files must exist in the cluster configuration directory. The cluster configuration Oracle directory object is created to access the cluster configuration files. Only the osak_admin_role is granted READ access to this directory. The cluster configuration files are readable only by the osak_admin_role. The cluster configuration files include the osakafka.properties file, and additional security files, such as SSL/TLS/PEM files and certificates. Keys and Certificates for SSL are stored in the Oracle keystore.

The cluster access Oracle directory object is used to control access to the Kafka cluster. This directory object does not contain any configuration files. Kafka sessions are exclusive to individual PDBs in the multitenant environment. Each PDB where you want to create an application to connect to a Kafka broker must create its own application.

No passwords must be embedded in files. Any embedded password properties in the osakafka.properties file will be ignored. All passwords must be stored as database credentials using the DBMS_CREDENTIAL package.

Kafka Clusters Using Kerberos Authentication

For Kafka clusters using Kerberos Authentication, the Kerberos ticket for the Kafka principal specified in the osakafka.properties file must be acquired on the database system, and renewed periodically outside of Oracle SQL access to Kafka.

The cluster configuration directory object and the cluster access directory object and database credential name must be supplied as input parameters to the DBMS_KAFKA_ADM.REGISTER_CLUSTER() call.

The Oracle SQL Access to Kafka administrator (a user with the osak_admin_role, the OSAK_ADMIN) performs the cluster registration and administration tasks.

Access to non-secure Kafka clusters requires that you create a cluster access database directory object to control access to the Kafka cluster. The grants on this database directory are used to control which Oracle Database users can access the Kafka cluster. This database directory has an empty path: it does not a need a corresponding operating system directory, and it also does not contain any files. Oracle recommends that the Oracle Directory Object Name for a cluster access database directory object takes the form OSAK_CLUSTER_NAME_ACCESS, where CLUSTER_NAME is the name of the Kafka cluster.

1. Create the cluster access database directory with an empty path. This directory is used to control which Oracle users can access the Kafka cluster.

   For example, create a cluster access database directory object called oaskaccess_kafkaclust1 with an empty path. This directory is used to control which Oracle users can access the Kafka cluster.

   SQL> CREATE DIRECTORY OSAK_KAFKACLUS2_ACCESS AS '';

2. On the target Oracle Database server, create the cluster configuration operating system directory in the Oracle base path directory, using the path Oracle_base/osak/cluster_name/config where Oracle_base is the Oracle base directory, and cluster_name is the Kafka cluster name. For example:

   mkdir /u01/app/oracle/osak/kafkaclus2/config

   Log in to the database as SYSDBA, start SQL, and create the corresponding Oracle directory object. In this example, the Kafka cluster name is KAFKACLUS2:

   SQL> CREATE DIRECTORY OSAK_KAFKACLUS2_CONFIG AS  'u01/app/oracle/osak/kafkaclus2/config';

3. Create an empty osakafka.properties file, or an osakafka.properties file with OSAK tuning or debugging properties.

4. In SQL, register the Kafka cluster using DBMS_KAKFA_ADM.REGISTER_CLUSTER(). For example, using the server hostname mykafkabootstrap-host, port 9092, for Kafka cluster KAFKACLUS2:

   SQL> select DBMS_KAFKA_ADM.REGISTER_CLUSTER (
             cluster_name  => ‘KAFKACLUS2’,
             bootstrap_servers =>‘Kafka-example-host:9092’, 
             kafka_provider => 
   DBMS_KAFKA_ADM.KAFKA_PROVIDER_APACHE, 
             cluster_access_dir => ‘OSAK_KAFKACLUS2_ACCESS’,
             credential_name => NULL,
             cluster_config_dir => ‘OSAK_KAFKACLUS2_CONFIG’,
              cluster_description =>  'My test cluster kafkaclus2’,
              options  => NULL) 
   from dual;

   If configuration is successful, then the registration return is 0 (zero):

   SQL> DBMS_KAFKA_ADM_RE…..
                0
   

5. Grant read access to a Kafka user. In the following example, user app2-usr is granted access to the Kafka cluster named KAFKACLUS2:

   SQL> grant read on directory osak_kafkaclus2_access to app2-usr;

Access to secure Kafka clusters requires configuration files, such as osakafka.properties, and additional security files such as SSL/TLS PEM files and certificates. These files are stored in a cluster configuration database directory object. The configuration files and directory are protected by the operating system directory and file access privileges,

The cluster configuration operating system directory is configured in the Oracle base directory, and is owned by the Oracle Installatoin owner, or Oracle user (oracle), and the Oracle Inventory Group (oinstall). The Oracle user and Oracle Inventory group must have directory privileges set to 750 (rwxr-x---) on and the osakafka.properties file in the directory must have privileges set to 540 (rw-r-----). All other files in the cluster configuration directory must have privileges set to and 440 (r--r-----).

1. Create a cluster access database directory object to control access to the Kafka cluster. The grants on this database directory object are used to control which Oracle Database users can access the Kafka cluster. This database directory has an empty path. That is, it does not need a corresponding operating system directory, and does not contain any files.

   For example, create a cluster access database directory object called oaskaccess_kafkaclust1 with an empty path. This directory is used to control which Oracle users can access the Kafka cluster. :

   SQL> CREATE DIRECTORY osakaccess_kafkaclus1 AS ''; 

2. On the target Oracle Database server, create the cluster configuration operating system directory in the Oracle base path directory, using the path Oracle_base/osak/cluster_name/config where Oracle_base is the Oracle base directory, and cluster_name is the Kafka cluster name. For example:

   mkdir /u01/app/oracle/osak/kafkaclus1/config

3. Log in to the database as SYSDBA, start SQL, and create the corresponding Oracle directory object in the target Oracle Database. Oracle recommends that you use OSAK_clustername_access for the database object name, where clusternamne is the name of the Kafka cluster. For example:

   CREATE DIRECTORY OSAK_KAFKACLUS1_CONFIG
      AS '/u01/app/oracle/osak/kafkaclus1/config';

4. Create the osakafka.properties file in the cluster configuration directory, based on the security protocol you use. This file is similar to librdkafka client properties file.

   In the following example, the osakafka.properties file is configured to use Secure Socket Layer (SSL) for the security protocol, with client authentication:

   security.protocol=ssl
   ssl.ca.location=ca-cert
   ssl.certificate.location=client_myhostname_client.pem 
   ssl.key.location=client_myhostname_client.key
   ssl.key.password=password-that-is-ignored

5. Copy the security files referred to by osakafka.properties into the cluster configuration directory. For example, where the ca-cert path is /etc/ssl/certs/:

   $cp /etc/ssl/certs/ca-cert /u01/app/oracle/osak/kafkaclus1/config;
   $cp /etc/ssl/certs/client-myhostname-client.pem /u01/app/oracle/osak/kafkaclus1/config;
   $cp /etc/ssl/certs/client-myhostname-client.key /u01/app/oracle/osak/kafkaclus1/config;
   

6. Set up credentials:

   • If you are using either SSL.key.location or sasl.username properties in the osakafka.properties file:

     Create a database credential to store the password required for authentication with the Kafka cluster using SSL SASI authentication. The corresponding password properties ssl.key.password or sasl.password are added automatically by DBMS_KAFKA during the cluster registration process. For example:

     begin
         dbms_credential.create_credential(
             credential_name => 'KAFKACLUS1CRED1',
             username => 'KAFKACLUS1',
             password => 'enter-ssl-key-password-or-sasl-password);
     end;
     /    

   • If your Kafka cluster uses GSSAPI/Kerberos as its authentication mechanism:

     Acquire the Kerberos ticket on the databases system for the Kafka principal listed in the osakafka.properties file

7. Log in as SYSDBA, start SQL, and register the Kafka cluster using the SYS.DBMS_KAFKA_ADM.REGISTER_CLUSTER() procedure. In the following example, the Kafka cluster KAFKACLUS1 is registered:

   select DBMS_KAFKA_ADM.REGISTER_CLUSTER ('KAFKACLUS1',
                'mykafkabootstrap-host:9092', 
                DBMS_KAFKA_ADM.KAFKA_PROVIDER_APACHE, 
                'OSAK_KAFKACLUS1_ACCESS'
                'KAFKACLUS1CRED1', 
                'OSAK_KAFKACLUS1_CONFIG',
                'My test cluster kafkaclus1') from dual;
   

   If successful, then the output should return 0 (zero). For example:

   SQL> DBMS_KAFKA_ADM_RE…..
                0
   

8. Grant read access to a Kafka user. In the following example, user app1-usr is granted access to the Kafka cluster named KAFKACLUS1:

   SQL> grant read on directory OSAK_KAFKACLUS1_ACCESS to app1-usr;