> Markdown version of [Clustering Support](https://vaadin.com/docs/v23/tools/collaboration/advanced/backend-api). Section index: [llms.txt](https://vaadin.com/docs/v23/tools/llms.txt)

# <a id="ce.backend"></a>Implementing Support for Clustered Environments

> **Important: Experimental feature**
>
> The Collaboration Kit backend for clustering support is an experimental feature. This means that its behavior, API, as well as its look and feel might change. You may enable it with a [Feature Flag](#feature-flag).

Using Collaboration Kit with its default settings in an application running in a clustered environment would result in users being able to collaborate only with others connected to the same application instance, that is, on the same node. To run clustered application deployments, Collaboration Kit provides the `Backend` superclass that can be extended to support such multi-instance environments.

This page shows how to implement a custom backend to support clustering based on the [Hazelcast](https://hazelcast.com) platform.

## <a id="understanding-the-concept-of-event-log"></a>Understanding the Concept of Event Log

Collaboration Kit uses a custom `Backend` implementation as the gateway for access to Event Logs. An Event Log is a strictly ordered log of submitted events involving Topic data, such as newly added items or value changes. The `EventLog` API provides methods to submit new events to the log and to add a subscriber to receive all past and future events. All events are marked by a unique identifier. With this, the API provides a method to remove all events in the log prior to a given identifier.

### <a id="implementing-an-event-log-for-hazelcast"></a>Implementing an Event Log for Hazelcast

Start by implementing the `EvengLog` interface for the reference Hazelcast platform. Hazelcast provides a straightforward streaming API based on shared maps and lists. The Event Log can be implemented to make use of a Hazelcast `IList`. However, first you need a class to store both the event identifier and payload.

`HazelcastBackend.java`

```java
private static final class IdAndPayload implements Serializable {

    private final UUID id;

    private final String payload;

    private IdAndPayload(UUID id, String payload) {
        this.id = id;
        this.payload = payload;
    }
}
```

Once you have that, you can start implementing the interface methods. The `submitEvent` uses the event identifier and payload, so you can store them in a new `IdAndPayload` object. You may also add it to the Hazelcast list for this Event Log.

`HazelcastBackend.java`

```java
public static class HazelcastEventLog implements EventLog {

    private final IList<IdAndPayload> list;
    public HazelcastEventLog(IList<IdAndPayload> list) {
        this.list = list;
    }

    @Override
    public void submitEvent(UUID id, String payload) {
        list.add(new IdAndPayload(id, payload));
    }

}
```

To implement subscriptions to events, add an item listener to the Hazelcast list. The subscriber will receive all past and future events for this Event Log. A newly added subscriber should receive initially all previous events in the log based on their original order. This allows it to catch up to the latest state. New events should be delivered, in order, only after all previous events have been delivered. The `subscribe` method may not be invoked again until the previous subscriber has been removed.

The `subscribe` method optionally takes the identifier of the last known event, so that the subscriber is notified only about newer events. If an identifier is provided, but not found in the Event Log, a `EventIdNotFoundException` should be thrown.

> **Tip: Exception handling**
>
> When the code calling this method catches the exception, it may re-attempt the subscription with another identifier.

`HazelcastBackend.java`

```java
@Override
public synchronized Registration subscribe(UUID newerThan,
        BiConsumer<UUID, String> eventSubscriber)
        throws EventIdNotFoundException {
    if (this.eventSubscriber != null) {
        throw new IllegalStateException(); // (1)
    }

    if (newerThan != null) {
        Optional<IdAndPayload> newerThanIdAndEvent = list.stream()
                .filter(item -> newerThan.equals(item.id)).findFirst();
        if (newerThanIdAndEvent.isEmpty()) {
            throw new EventIdNotFoundException(
                    "newerThan doesn't " + "exist in the log."); // (2)
        }
    }
    this.newerThan = newerThan;
    this.eventSubscriber = eventSubscriber;
    nextEventIndex = 0;

    UUID registrationId = list
            .addItemListener(new ItemListener<IdAndPayload>() {
                @Override
                public void itemAdded(ItemEvent<IdAndPayload> item) {
                    deliverEvents();
                }

                @Override
                public void itemRemoved(ItemEvent<IdAndPayload> item) {
                    handleRemoveItem();
                }
            }, false); // (3)

    // Deliver initial events
    deliverEvents(); // (4)

    return () -> {
        synchronized (this) {
            list.removeItemListener(registrationId);
            this.eventSubscriber = null;
        }
    }; // (5)
}
```

1. Only a single subscriber is allowed, so an exception is thrown if one is already set.

2. If an event identifier is provided, the list is checked to see if it exists in it. If it doesn’t, an `EventIdNotFoundException` is thrown.

3. An item listener is added to the Hazelcast list to handle new items and ones that have been removed.

4. Then all past events are initially delivered.

5. Finally, `Registration` is returned that can be used to remove the subscriber.

The `deliverEvents` method is `synchronized` to prevent it from being invoked by multiple threads simultaneously — and to avoid duplicate events being notified to the subscriber. The method keeps track of the Hazelcast list index to identify the next event and it increments that index until all of the events are delivered. If an event identifier has been set as the starting point, no events are delivered until that identifier is reached.

`HazelcastBackend.java`

```java
private synchronized void deliverEvents() {
    while (nextEventIndex < list.size()) {
        IdAndPayload event = list.get(nextEventIndex++);
        if (this.newerThan == null) {
            eventSubscriber.accept(event.id, event.payload);
        } else {
            if (event.id.equals(newerThan)) {
                this.newerThan = null;
            }
        }
    }
}
```

Finally, the last method to implement for the `EventLog` interface is the `truncate` method. This method is used to limit the number of events contained in the log, preventing it from growing infinitely. It takes the identifier of the oldest known event that should be preserved, or, if a `null` identifier is provided, it empties the entire log.

To implement this behavior for Hazelcast, create a simple `Predicate` and pass it to the list’s `removeIf` method.

`HazelcastBackend.java`

```java
@Override
public synchronized void truncate(UUID olderThan) {
    Predicate<IdAndPayload> filter = e -> true;
    if (olderThan != null) {
        Optional<IdAndPayload> olderThanEvent = list.stream()
                .filter(item -> olderThan.equals(item.id)).findFirst();
        if (olderThanEvent.isEmpty()) {
            // NOOP
            return;
        }
        filter = new Predicate<>() {
            boolean found;

            @Override
            public boolean test(IdAndPayload event) {
                found = found || olderThan.equals(event.id);
                return !found;
            }
        };
    }
    list.removeIf(filter);
}
```

## <a id="opening-an-event-log"></a>Opening an Event Log

Now that you have a Hazelcast implementation of the `EventLog` interface, to be able to create and get instances of it, you need to extend the `Backend` class. Since this implementation depends only on a single Hazelcast `IList`, it’s very easy to implement the `openEventLog` method. You would do so by returning a new instance of the `HazelcastEventLog` with the list named after the `logId` parameter.

`HazelcastBackend.java`

```java
@Override
public EventLog openEventLog(String logId) {
    return new HazelcastEventLog(hz.getList(logId));
}
```

## <a id="managing-event-log-snapshots"></a>Managing Event Log Snapshots

A *snapshot* is an opaque representation of data at a certain moment in time. It can be used, for example, by nodes joining a cluster to catch-up quickly with a recent state of data, without the need to replay all of the events from the beginning. Snapshots are identified by name. Each version of a named snapshot is assigned an additional unique identifier.

### <a id="loading-a-snapshot"></a>Loading a Snapshot

To load the latest version of a snapshot, the `Backend` class provides the `loadLatestSnapshot` method. This method can be implemented for Hazelcast by using a map to store the latest available snapshot.

`HazelcastBackend.java`

```java
@Override
public CompletableFuture<Snapshot> loadLatestSnapshot(String name) {
    return CompletableFuture.completedFuture(snapshots.get(name));
}
```

### <a id="submitting-a-new-snapshot"></a>Submitting a New Snapshot

To submit a new snapshot of data, the `replaceSnapshot` method should be used. It takes the name of the snapshot, the expected unique identifier of the latest snapshot, the unique identifier of the new snapshot and the payload of the snapshot itself. To implement this method for Hazelcast, you need some logic to verify that the latest snapshot is the expected one.

`HazelcastBackend.java`

```java
@Override
public CompletableFuture<Void> replaceSnapshot(String name, UUID expectedId,
        UUID newId, String payload) {
    Snapshot currentSnapshot = snapshots.computeIfAbsent(name,
            k -> new Snapshot(null, null));

    if (Objects.equals(expectedId, currentSnapshot.getId())) {
        Snapshot idAndPayload = new Snapshot(newId, payload);
        snapshots.put(name, idAndPayload);
    }

    return CompletableFuture.completedFuture(null);
}
```

## <a id="node-identity-and-membership-events"></a>Node Identity and Membership Events

The primary purpose of the Backend API is to support collaboration in applications deployed in clustered environments. Every `Backend` instance represents a member of the cluster and is uniquely identified by a `UUID`, which should be returned by the `getNodeId` method. For your Hazelcast implementation, return the local member identifier.

`HazelcastBackend.java`

```java
@Override
public UUID getNodeId() {
    return hz.getCluster().getLocalMember().getUuid();
}
```

When many `Backend` instances are involved, it’s necessary to know when they join or leave the cluster. For this purpose, the `Backend` should provide an implementation of the `addMembershipListener` method that takes a `MembershipListener` and notifies it when cluster members join or leave. Since Hazelcast itself uses the same concept, the implementation is straightforward: you only need to map Hazelcast’s events to Collaboration Kit’s `MembershipEvent` events. This takes the `MembershipEventType` (`JOIN` or `LEAVE`) and the identifier of the member.

`HazelcastBackend.java`

```java
@Override
public Registration addMembershipListener(
        MembershipListener membershipListener) {
    UUID registrationId = hz.getCluster()
            .addMembershipListener(new InitialMembershipListener() {

                @Override
                public void init(InitialMembershipEvent event) {
                    event.getMembers()
                            .forEach(member -> submitEvent(
                                    MembershipEventType.JOIN,
                                    member.getUuid()));
                }

                @Override
                public void memberAdded(MembershipEvent membershipEvent) {
                    submitEvent(MembershipEventType.JOIN,
                            membershipEvent.getMember().getUuid());
                }

                @Override
                public void memberRemoved(MembershipEvent membershipEvent) {
                    submitEvent(MembershipEventType.LEAVE,
                            membershipEvent.getMember().getUuid());
                }

                private void submitEvent(MembershipEventType type,
                        UUID id) {
                    membershipListener.handleMembershipEvent(
                            new com.vaadin.collaborationengine.MembershipEvent(
                                    type, id, getCollaborationEngine()));
                }
            });
    return () -> hz.getCluster().removeMembershipListener(registrationId);
}
```

> **Note: Returning a registration object**
>
> The `addMembershipListener` should return a `Registration` object that can be used later to remove the listener.

## <a id="feature-flag"></a>Feature Flag

To use the Collaboration Kit backend for clustering support, you must enable it with the `collaborationEngineBackend` feature flag.

See [Feature Flags](https://vaadin.com/docs/v23/configuration/feature-flags.md) for more information.

`AB472607-53E3-481D-AF99-93E3F6ED8B61`