Producer APIs, allow Kafka application to publish messages into Oracle Transaction Event Queues. Kafka application needs to provide Oracle specific properties which are oracle.host, oracle.port, oracle.servicename, and oracle.instancename. More details about this properties are mentioned in configuration section. These properties are used to setup the database connection and produce the message into TEQ. In the current release, Oracle's implementation of KafkaProducer supports only a subset of the APIs.

Internally, Oracle Kafka Producer object encapsulates AQJMS producer object which is used to publish messages into Oracle TEQ. Similar to Kafka Producer, Producer also stores messages in batches. Each send() call will append a Kafka Record into a particular batch based on its Topic and Partition. A background thread will publish the entire batch, one at a time into Oracle TEQ. Each batch publish is committed by Producer. In current release, a topic can have only one partition and hence all KafkaRecords will be published into a single partition of TEQ.

The following KafkaProducer APIs are supported in Oracle Database 20c.

• Constructor:

  KafkaProducer: Creates a producer object and internal AQ JMS objects. KafkaProducer class has four types of constructors defined, which all takes configuration parameters as input.

• Methods:

  • send(ProducerRecord) , send(ProducerRecord, Callback):

    The send method asynchronously publishes a message into TEQ. This method returns immediately once a Kafka Record has been stored in the buffer of records waiting to be sent. If buffer memory is full, then send call blocks for a maximum of time max.block.ms. This allows sending many records in parallel without blocking to wait for the response after each one. Records will be published into the topic using AQ JMS.

    The result of the send is a Future<RecordMetadata> specifying the partition the record was sent to, the offset it was assigned and the timestamp of the record. Both the version send(ProducerRecord) and send(ProducerRecord, Callback) will be supported.

  • close: Closes the producer, its sender thread and frees the accumulator. It also closes internal AQ JMS objects like connection, session JMS producer and so on.

• Classes

  • ProducerRecord: A class that represents a message in Kafka platform. It is translated into a message for TEQ platform, namely, AQ JMS message. Relevant fields like Payload and Key can be directly translated into TEQ payload and message key for TEQ.

  • RecordMetadata: This contains metadata of the record like topic, partition, offset, timestamp etc of the Record in KafkaPlatform. This is assigned value relevant for TEQs. A message id of TEQ is converted into an offset of RecordMetadata.

  • Callback Interface: A callback function which is executed once a Record is successfully published into Kafka topic.

  • Partitioner Interface: Defines methods which maps a Key of the message to a partition number of the topic. A partition number is analogous to a stream id of TEQs.

• Properties

  • Key Serializer and Value Serializer: Converts Key and payload into byte array respectively. The Accumulator module will store the payloads in the form of byte array. Sender thread will then form an AQjmsBytes message and publish the message using AQ JMS Array Enqueue API.

  • acks: For okafka, only value relevant for acks property is all. Any other field set by the user is ignored.

  • linger.ms: Time in miliseconds for which sender thread waits before publishing the records in TEQ.

  • batch.size: Total size of records to be batched in bytes for which sender thread waits before publishing records in TEQ.

  • buffer.memory: Total memory in bytes the accumulator can hold.

  • max.block.ms: If buffer.memory size is full in accumulator, then wait for max.block.ms amount of time before send() method can receive out of memory error.

  • retries: This property enables producer to resend the record in case of transient errors. This value is an upper limit on how many resends.

  • retry.backoff.ms : The amount of time to wait before attempting to retry a failed request to a given topic partition. This avoids repeatedly sending requests in a tight loop under some failure scenarios

  • bootstrap.servers: IP address and port of a machine where database instance running.

Consumer API allows applications to read streams of data from Transactional Event Queue. Kafka consumer for TEQ uses AQ JMS APIs and use JDBC driver to consume messages from Oracle TEQ. For Oracle Kafka, consuming message from a topic implies dequeuing messages from Transactional Event Queue.

Similar to Kafka, in TEQ's implementation , a consumer group contains many consumer instances. Each consumer group has a unique group-id. Each consumer internally maintains a single connection/session to Oracle Database instance provided by bootstrap.servers property. For this release, since a topic can have only one partition, only one of the consumer instances will be assigned this single partition. A partition once assigned to a consumer of a consumer group then remains with that consumer till the session is closed. No two consumers from same group are assigned same partition of a topic.

The following KafkaConsumer APIs are supported in Oracle Database 20c.

• Constructor: KafkaConsumer: Creates a consumer that allows the application to consume message from key based TEQ. Internal client side TEQ objects created are not visible to client application. All variation of the KafkaConsumer constructor are supported in Oracle Database 20c.

• Methods:

  • Subscribe: This method takes a list of topics to subscribe to. In Oracle Database 20c, only the first topic of the list will be subscribed to. An exception is thrown if size of list is > 1. This method creates a durable subscriber on TEQ server side with Group-Id as subscriber name.

  • Poll: poll method returns a batch of messages from the assigned partition from TEQ. It attempts to dequeue a message from the key based TEQ for the subscriber. TEQ uses array dequeue API of AQ JMS to receive a batch of messages dequeued from the queue. The size of the batch depends on the parameter max.poll.records set by the kafka client application. Poll takes time in milliseconds as an argument. AQ JMS API of array dequeue can pass this timeout time as a dequeue option to the TEQ Server and make dequeue call wait for messages till the timeout time, if the full array batch is not complete.

    When poll is invoked for the first time, Oracle TEQ assigns a single available partition to this Kafka consumer. This assignment stays for the entire lifetime of the Kafka consumer. Messages returned belongs to the partition assigned to the consumer. One queue partition is assigned to each Kafka consumer. It is the responsibility of the application developer to start as many consumers as number of partitions of the queue. If the number of Kafka consumers are less than the number of partitions, then messages from unassigned partitions are never consumed. If the number of Kafka consumers are more than the number of partitions, then extra consumers will not be assigned any partition and hence, will not be able to consume any messages. No two consumer application will consume from same partition at the same time.

  • commitSync: Commits all consumed messages. Commit to an offset is not supported in Oracle Database 20c. This call directly calls commit on the database which commits all consumed messages from TEQ.

  • commitAsync: This call is translated into commitSync. A callback function passed as argument gets executed once the commit is successful.

  • Unsubscribe: Unsubscribes the topic that it has subscribed to. A consumer can no longer consume messages from unsubscribed topics. This call does not remove a subscriber group from the TEQ metadata. Other consumer application can still continue to consume.

  • close: Closes the consumer and unsubscribes the topic it has subscribed to.

• Class: ConsumerRecord: A class that represents a consumed record in Kafka platform. In Oracle Dataase 20c, AQ JMS message is converted into ConsumerRecord.

• Properties:

  • key.deserializer and value.deserialzer: In Oracle TEQ's Key based partitioned queue key, value are stored as byte array in user property, payload of JMS message respectively. On consuming these byte arrays are deserialized into key, value having user provided format internally by the consumer using key.deserialize and value.deserializer respectively.

  • group.id: This is a consumer group name for which messages are consumed from the Kafka topic. This property is used as a durable subscriber name for key based TEQs.

  • max.poll.records: Maximum number of records to fetch in single array dequeue call from an Oracle TEQ server.

  • fetch.max.wait.ms: Maximum amount of time in milliseconds to wait for fetching messages if not available.

  • enable.auto.commit: Enables auto commit of consumed messages for every specified interval.

  • auto.commit.interval.ms: Interval in milliseconds for auto commit of messages.

  • bootstrap.servers: IP address and port of a machine where database instance running.