Developing With Kafka APIs

<dependency>
    <groupId>com.oracle.database.messaging</groupId>
    <artifactId>okafka</artifactId>
    <version>${okafka.version}</version>
</dependency>

implementation "com.oracle.database.messaging:okafka:${okafkaVersion}"

user = <database username>
password = <database password>

Properties props = new Properties();
// Database service name
props.put("oracle.service.name", "freepdb1");
// Connection protocol. Set to either PLAINTEXT or SSL
props.put("security.protocol", "PLAINTEXT");
// Oracle Database hostname and (port)
props.put("bootstrap.servers", "localhost:1521");
// Path to the directory containing ojdbc.properties
props.put("oracle.net.tns_admin", "<ojdbc.properties directory>");

Properties props = new Properties();
// Database service name
props.put("oracle.service.name", "mypdb");
// Connection protocol. Set to either PLAINTEXT or SSL
props.put("security.protocol", "SSL");
// Oracle Database hostname and (port)
props.put("bootstrap.servers", "database_hostname");
// Path to wallet directory
props.put("oracle.net.tns_admin", "<wallet directory>");

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.errors.TopicExistsException;

// Implements org.apache.kafka.clients.admin.Admin for Transactional Event Queues
import org.oracle.okafka.clients.admin.AdminClient;

public class TopicCreator {
    public static void createTopic(Properties props, String topicName, int partitions) {
        NewTopic newTopic = new NewTopic(topicName, partitions, (short) 1);
        try (Admin admin = AdminClient.create(props)) {
            admin.createTopics(Collections.singletonList(newTopic))
                    .all()
                    .get();
        } catch (ExecutionException | InterruptedException e) {
            // Handle case where topic already exists or handle other exceptions as appropriate.
            if (e.getCause() instanceof TopicExistsException) {
                System.out.println("Topic already exists, skipping creation");
            } else {
                throw new RuntimeException(e);
            }
        }
    }
}

// Assume props is a configured Properties object for Oracle Database
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// Configure additional properties as needed.
props.put("enable.idempotence", "true");

// Create the producer
Producer<String, String> producer = new KafkaProducer<>(props);
// Send a message
producer.send(new ProducerRecord<>("my_topic", "my first message!"));

// Assume props is already configured with Oracle Database authentication properties as detailed earlier.
props.put("group.id", "MY_CONSUMER_GROUP");
props.put("enable.auto.commit", "false");
props.put("max.poll.records", 2000);
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");

// Create the consumer
Consumer<String, String> consumer = new KafkaConsumer<>(props);
// Subscribe to topics
consumer.subscribe(List.of("my_topic"));
// Poll for records. Typically done in a loop
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));

<dependency>
    <groupId>com.oracle.database.spring</groupId>
    <artifactId>oracle-spring-boot-starter-json-collections</artifactId>
    <version>${oracle-starters.version}</version>
</dependency>

implementation "com.oracle.database.spring:oracle-spring-boot-starter-json-collections:${oracleStartersVersion}"

package com.example;

import com.oracle.spring.json.jsonb.JSONB;
import org.apache.kafka.common.serialization.Serializer;

/**
 * The JSONBSerializer converts java objects to a JSONB byte array.
 * @param <T> serialization type.
 */
public class JSONBSerializer<T> implements Serializer<T> {
    private final JSONB jsonb;

    public JSONBSerializer(JSONB jsonb) {
        this.jsonb = jsonb;
    }

    @Override
    public byte[] serialize(String s, T obj) {
        return jsonb.toOSON(obj);
    }
}

JSONB json = new JSONB(new OracleJsonFactory(), (YassonJsonb) JsonbBuilder.create());
Serializer<MyPOJO> keySerializer = new StringSerializer();
Serializer<MyPOJO> valueSerializer = new JSONBSerializer<>(jsonb);
Producer<String, MyPOJO> producer = KafkaProducer<>(props, keySerializer, valueSerializer);

package com.example;

import java.nio.ByteBuffer;

import com.oracle.spring.json.jsonb.JSONB;
import org.apache.kafka.common.serialization.Deserializer;

/**
 * The JSONBDeserializer converts JSONB byte arrays to java objects.
 * @param <T> deserialization type
 */
public class JSONBDeserializer<T> implements Deserializer<T> {
    private final JSONB jsonb;
    private final Class<T> clazz;

    public JSONBDeserializer(JSONB jsonb, Class<T> clazz) {
        this.jsonb = jsonb;
        this.clazz = clazz;
    }

    @Override
    public T deserialize(String s, byte[] bytes) {
        return jsonb.fromOSON(ByteBuffer.wrap(bytes), clazz);
    }
}

JSONB json = new JSONB(new OracleJsonFactory(), (YassonJsonb) JsonbBuilder.create());
Deserializer<MyPOJO> keyDeserializer = new StringDeserializer();
Deserializer<MyPOJO> valueDeserializer = new JSONBDeserializer<>(jsonb, MyPOJO.class);
Consumer<String, MyPOJO> consumer = new KafkaConsumer<>(props, keyDeserializer, valueDeserializer);