[[cloudevents_integration_module]]
= CloudEvents Integration Module (Experimental)

{brandname} version 12 introduces an experimental module that converts
{brandname} events to CloudEvents events and sends them to a Kafka topic
in structured mode, with the JSON format.

This allows Infinispan be further used as a Knative source.

There are two broad kinds of events, and they can be sent to different
Kafka topics:

* Cache entry modifications: created, updated, removed, expired
* Audit events: user login, access denied


== Configuration

CloudEvents integration is enabled by adding the module to the global
configuration and configuring at least a list of bootstrap Kafka servers
and a topic.

[source,xml,options="nowrap",subs=attributes+]
----
<?xml version="1.0" encoding="UTF-8"?>
<infinispan
xmlns="urn:infinispan:config:{schemaversion}"
xmlns:ce="urn:infinispan:config:cloudevents:{schemaversion}"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:infinispan:config:{schemaversion}
https://infinispan.org/schemas/infinispan-config-{schemaversion}.xsd
urn:infinispan:config:cloudevents:{schemaversion}
https://infinispan.org/schemas/infinispan-cloudevents-config-{schemaversion}.xsd">

    <cache-container default-cache="default">
        <transport/>
        <ce:cloudevents bootstrap-servers="127.0.0.1:9092"
                                   audit-topic="audit"
                                   cache-entries-topic="cache-events"/>

        <replicated-cache name="default">
            <ce:cloudevents-cache enabled="false"/>
        </replicated-cache>
    </cache-container>

</infinispan>
----

[source,java,options="nowrap",subs=attributes+]
----
GlobalConfigurationBuilder managerBuilder = GlobalConfigurationBuilder.defaultClusteredBuilder();
CloudEventsGlobalConfigurationBuilder cloudEventsGlobalBuilder =
    managerBuilder.addModule(CloudEventsGlobalConfigurationBuilder.class);
cloudEventsGlobalBuilder.bootstrapServers("localhost:9092");
cloudEventsGlobalBuilder.cacheEntriesTopic("ispn");
----

Currently only the list of bootstrap servers, the number of acks, and the
cache entries/audit topic names are configurable.


== Event Format

The events are sent in
link:https://github.com/cloudevents/spec/blob/v1.0/kafka-protocol-binding.md#13-content-modes[Kafka structured mode],
in the
link:https://github.com/cloudevents/spec/blob/v1.0/json-format.md[JSON event format]

This is an example of an event:

[source,json,options="nowrap",subs=attributes+]
----
{
    "specversion": "1.0",
    "source": "/infinispan/CLUSTER/testCache",
    "type": "org.infinispan.entry.created",
    "time": "2020-10-29T22:05:08.767950Z",
    "subject": "key-1",
    "id": "key-1:CommandInvocation:Test-NodeA:0",
    "data": {
       "property": "value"
    }
}
----

The `source` field starts with `/infinispan/`, then the cluster name,
and ends with the cache name.

The `type` field is a straightforward mapping of the {brandname} cache entry
event types, prefixed with `org.infinispan.entry`.

The `data` field is the new entry value (or the old value, for remove events).
If the value is Protostream-encoded, a Java object that can be marshalled
to Protostream, or a Java primitive wrapper, it is converted to JSON.
If the value is not encoded with Protostream or it is a Java object
and the cache manager is configured to use another marshaller,
the marshalled value is written as a string.
If the marshalled value is not a valid UTF-8 string, it is first Base64-encoded.

The `subject` field is the affected cache key, converted to a string
using the same mechanism as the values.
The only difference is that the resulting JSON is also wrapped in a string.

TODO: The JSON conversion currently adds `\n` and space characters,
which are escaped and preserved in the `subject` field value.
We may want to Base64-encode protostream keys instead.

The `id` field is composed from the key and a transaction/invocation id.
Expiration events do not have an invocation id, so a random id is generated
instead.
