Lab 4.3: Build It — A Custom ClusterStateListener / Custom Metadata

You have read the publication and applier model (Lab 4.2). Now you build against it. In this lab you write a plugin that registers a ClusterStateListener reacting to real cluster state changes — it logs whenever an index is created or deleted — and, as a stretch, registers a custom Metadata.Custom type that rides inside the cluster state and survives serialization. You then test it with the OpenSearch test framework (OpenSearchSingleNodeTestCase / OpenSearchIntegTestCase).

This is the smallest realistic way to hook the coordination layer, and it is exactly how ecosystem plugins (index-management, security, cross-cluster-replication) observe and extend cluster state.

Background

Two extension points from Lab 4.2:

• ClusterStateListener — clusterChanged(ClusterChangedEvent event) is called on every node after a new state is applied. The ClusterChangedEvent carries old state, new state, and helpers like indicesCreated() / indicesDeleted().
• Metadata.Custom — a plugin-defined, named, Writeable + ToXContent blob that lives inside ClusterState.metadata(), is published with the state, and is persisted. Registering one requires entries in NamedWriteableRegistry (transport) and NamedXContentRegistry (XContent), which a Plugin contributes via getNamedWriteables() / getNamedXContent().

Wiring is via ClusterService.addListener(...). You get the ClusterService in Plugin.createComponents(...).

Deep-dive companions: cluster-state.md, cluster-state-publishing.md, plugin-internals.md. The applier/listener distinction is from Lab 4.2; plugin/action mechanics are from Lab 3.3.

Why This Lab Matters for Contributors

Reacting to cluster state is one of the most common things a real OpenSearch plugin does: enforce a policy when an index appears, clean up resources when one is deleted, replicate metadata, track state. Custom Metadata is how features store cluster-scoped data (ISM policies, security config, replication bookkeeping). If you can register a listener and a custom metadata type and prove they survive serialization in a test, you've done the core of a stateful plugin.

Prerequisites

• You completed Lab 3.3 (you can build and install a plugin).
• You read Lab 4.2 (appliers vs listeners; two-phase publish).
• A standalone plugin project skeleton like Lab 3.3's, or work inside a modules/ sandbox module.

Note: Method signatures on Plugin (createComponents, getNamedWriteables, getNamedXContent) and on Metadata.Custom have evolved across versions. Where a signature doesn't match your branch, copy the current one from an in-tree user: grep -rln "implements Metadata.Custom\|getNamedWriteables\|createComponents" server/ modules/ plugins/ | head and mirror it. Package paths for Writeable/ToXContent/StreamInput may be under org.opensearch.core.* — verify with grep -rn "class StreamInput" libs/.

Part A — The ClusterStateListener (45 min)

Step 1 (10 min) — The listener

package org.example.statewatch;

import org.opensearch.cluster.ClusterChangedEvent;
import org.opensearch.cluster.ClusterStateListener;
import org.opensearch.core.index.Index;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

public class IndexLifecycleLogger implements ClusterStateListener {

    private static final Logger logger = LogManager.getLogger(IndexLifecycleLogger.class);

    @Override
    public void clusterChanged(ClusterChangedEvent event) {
        // indicesCreated() / indicesDeleted() are diffed for us from old vs new state.
        for (String created : event.indicesCreated()) {
            logger.info("index created: [{}] (cluster state v{})",
                created, event.state().version());
        }
        for (Index deleted : event.indicesDeleted()) {
            logger.info("index deleted: [{}] (cluster state v{})",
                deleted.getName(), event.state().version());
        }
    }
}

Notes:

• clusterChanged runs on the single applier thread after appliers. It must be fast and non-blocking (Lab 4.2 warning). Logging is fine; calling out to the network is not.
• ClusterChangedEvent already computed the create/delete deltas from old vs new state — you don't diff the metadata yourself.

Step 2 (10 min) — A component to register the listener

You need the ClusterService to call addListener. The plugin gets it in createComponents. Make a small component that registers and unregisters the listener cleanly.

package org.example.statewatch;

import org.opensearch.cluster.service.ClusterService;
import org.opensearch.common.lifecycle.AbstractLifecycleComponent;

public class IndexLifecycleService extends AbstractLifecycleComponent {

    private final ClusterService clusterService;
    private final IndexLifecycleLogger listener = new IndexLifecycleLogger();

    public IndexLifecycleService(ClusterService clusterService) {
        this.clusterService = clusterService;
    }

    @Override
    protected void doStart() {
        clusterService.addListener(listener);
    }

    @Override
    protected void doStop() {
        clusterService.removeListener(listener);
    }

    @Override
    protected void doClose() {
        // nothing else to release
    }
}

Step 3 (10 min) — The plugin: wire it in createComponents

package org.example.statewatch;

import org.opensearch.cluster.service.ClusterService;
import org.opensearch.common.lifecycle.LifecycleComponent;
import org.opensearch.plugins.Plugin;
import org.opensearch.plugins.ClusterPlugin;
import org.opensearch.core.common.io.stream.NamedWriteableRegistry;

import java.util.Collection;
import java.util.Collections;
import java.util.List;

// ... createComponents has many parameters; trim to what's needed and match your branch's signature.
public class StateWatchPlugin extends Plugin implements ClusterPlugin {

    @Override
    public Collection<Object> createComponents(/* PluginServices services -- see note */ Object... services) {
        // On your branch, createComponents takes a long parameter list (Client, ClusterService,
        // ThreadPool, ...). Pull ClusterService out of it and construct the service.
        // The illustrative body:
        //
        //   ClusterService clusterService = ...; // from the parameters
        //   return Collections.singletonList(new IndexLifecycleService(clusterService));
        //
        // Returning a LifecycleComponent means OpenSearch will start()/stop() it for you,
        // which is what registers/unregisters the listener.
        throw new UnsupportedOperationException("fill from your branch's createComponents signature");
    }
}

Note — createComponents signature: It is the most version-sensitive method on Plugin. On current main it takes a single PluginServices-style bundle or a long parameter list (Client client, ClusterService clusterService, ThreadPool threadPool, ResourceWatcherService ..., ScriptService ..., NamedXContentRegistry ..., Environment ..., NodeEnvironment ..., NamedWriteableRegistry ..., IndexNameExpressionResolver ..., Supplier<RepositoriesService> ...). Copy the exact list: grep -rn "createComponents" server/src/main/java/org/opensearch/plugins/Plugin.java Then return List.of(new IndexLifecycleService(clusterService)). Returning the component as a LifecycleComponent is what gets doStart() (and thus addListener) called.

The concrete, working body once you've pulled ClusterService from the parameters:

ClusterService clusterService = /* the ClusterService argument */;
return Collections.singletonList(new IndexLifecycleService(clusterService));

Step 4 (15 min) — Test it with OpenSearchSingleNodeTestCase

A single-node test is the fastest way to prove the listener fires. It loads your plugin, creates an index, and asserts the listener saw it.

package org.example.statewatch;

import org.opensearch.plugins.Plugin;
import org.opensearch.test.OpenSearchSingleNodeTestCase;

import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class IndexLifecycleListenerIT extends OpenSearchSingleNodeTestCase {

    @Override
    protected Collection<Class<? extends Plugin>> getPlugins() {
        return Collections.singletonList(StateWatchPlugin.class);
    }

    public void testListenerFiresOnIndexCreate() throws Exception {
        // Register our own listener directly on the node's ClusterService to observe the event
        // deterministically (the plugin's logger is fine for manual runs, but a latch is testable).
        CountDownLatch created = new CountDownLatch(1);
        getInstanceFromNode(org.opensearch.cluster.service.ClusterService.class)
            .addListener(event -> {
                if (event.indicesCreated().contains("watch-me")) {
                    created.countDown();
                }
            });

        client().admin().indices().prepareCreate("watch-me").get();

        assertTrue("listener should have observed index creation",
            created.await(10, TimeUnit.SECONDS));
    }
}

Run it:

# If inside the OpenSearch tree as a module/sandbox:
./gradlew :modules:<your-module>:test --tests "*IndexLifecycleListenerIT"
# Standalone plugin project:
./gradlew test --tests "*IndexLifecycleListenerIT"

Note: getInstanceFromNode(...) reaches into the test node's Guice injector for a real service instance — the supported way to inspect a node's internals in OpenSearchSingleNodeTestCase. The CountDownLatch makes the assertion deterministic instead of relying on log scraping. The test framework is the subject of Level 5.

Part B (Stretch) — A custom Metadata.Custom type (45 min)

This is the harder, optional half: a plugin-defined blob that lives in the cluster state, publishes with it, and round-trips through transport and XContent.

Step 5 (15 min) — The custom metadata

package org.example.statewatch;

import org.opensearch.cluster.AbstractNamedDiffable;
import org.opensearch.cluster.NamedDiff;
import org.opensearch.cluster.metadata.Metadata;
import org.opensearch.core.common.io.stream.StreamInput;
import org.opensearch.core.common.io.stream.StreamOutput;
import org.opensearch.core.xcontent.XContentBuilder;
import org.opensearch.Version;

import java.io.IOException;
import java.util.EnumSet;

public class WatchMetadata extends AbstractNamedDiffable<Metadata.Custom> implements Metadata.Custom {

    public static final String TYPE = "watch_metadata";

    private final long createdCount;

    public WatchMetadata(long createdCount) {
        this.createdCount = createdCount;
    }

    public WatchMetadata(StreamInput in) throws IOException {
        this.createdCount = in.readVLong();
    }

    @Override
    public String getWriteableName() {
        return TYPE;
    }

    @Override
    public Version getMinimalSupportedVersion() {
        return Version.CURRENT.minimumIndexCompatibilityVersion();
    }

    @Override
    public void writeTo(StreamOutput out) throws IOException {
        out.writeVLong(createdCount);
    }

    @Override
    public EnumSet<Metadata.XContentContext> context() {
        // Persist in snapshots + on-disk state; include in the API. Choose per feature.
        return Metadata.ALL_CONTEXTS;
    }

    @Override
    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
        builder.field("created_count", createdCount);
        return builder;
    }

    public long getCreatedCount() {
        return createdCount;
    }

    public static NamedDiff<Metadata.Custom> readDiffFrom(StreamInput in) throws IOException {
        return readDiffFrom(Metadata.Custom.class, TYPE, in);
    }
}

Step 6 (15 min) — Register the named writeable + named XContent

The cluster state is serialized by name, so a custom type must be registered in both registries. A Plugin contributes these:

@Override
public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
    return List.of(
        new NamedWriteableRegistry.Entry(Metadata.Custom.class, WatchMetadata.TYPE, WatchMetadata::new),
        new NamedWriteableRegistry.Entry(NamedDiff.class, WatchMetadata.TYPE, WatchMetadata::readDiffFrom)
    );
}

@Override
public List<NamedXContentRegistry.Entry> getNamedXContent() {
    return List.of(
        new NamedXContentRegistry.Entry(
            Metadata.Custom.class,
            new ParseField(WatchMetadata.TYPE),
            WatchMetadata::fromXContent   // implement a fromXContent parser to match toXContent
        )
    );
}

Warning: If you write a custom metadata to the cluster state but forget the NamedWriteableRegistry entry, the next state publication will fail to deserialize on receiving nodes (IllegalArgumentException: Unknown NamedWriteable [...]) and the cluster will be unable to apply state — a cluster-wide outage from one missing registration. This is the classic custom metadata bug. Always wire both registries, and always add a serialization round-trip test.

Step 7 (15 min) — Round-trip test for the custom metadata

Use an AbstractNamedWriteableTestCase (or AbstractSerializingTestCase / AbstractWireSerializingTestCase) to prove the type survives transport serialization:

package org.example.statewatch;

import org.opensearch.cluster.metadata.Metadata;
import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
import org.opensearch.core.common.io.stream.Writeable;
import org.opensearch.test.AbstractNamedWriteableTestCase;

import java.util.List;

public class WatchMetadataTests extends AbstractNamedWriteableTestCase<Metadata.Custom> {

    @Override
    protected WatchMetadata createTestInstance() {
        return new WatchMetadata(randomNonNegativeLong());
    }

    @Override
    protected Class<Metadata.Custom> categoryClass() {
        return Metadata.Custom.class;
    }

    @Override
    protected NamedWriteableRegistry getNamedWriteableRegistry() {
        return new NamedWriteableRegistry(List.of(
            new NamedWriteableRegistry.Entry(Metadata.Custom.class, WatchMetadata.TYPE, WatchMetadata::new)
        ));
    }

    @Override
    protected Writeable.Reader<Metadata.Custom> instanceReader() {
        return WatchMetadata::new;
    }
}

./gradlew test --tests "*WatchMetadataTests"

This round-trips a random WatchMetadata through StreamOutput→StreamInput via the registry and asserts equality — the exact protection against the "unknown NamedWriteable" outage above. Wire testing is the heart of Level 5 and BWC (Level 9).

Implementation Requirements

• IndexLifecycleLogger implements ClusterStateListener, uses indicesCreated()/indicesDeleted(), and does no blocking work.
• The listener is registered via ClusterService.addListener(...) from a LifecycleComponent returned by createComponents(...), and unregistered on stop.
• An OpenSearchSingleNodeTestCase proves the listener fires on index creation (deterministically, via a latch).
• (Stretch) WatchMetadata implements Metadata.Custom, is registered in both getNamedWriteables() and getNamedXContent(), and a round-trip serialization test passes.

Troubleshooting

Expected Output

Listener test output (abridged):

> Task :test
IndexLifecycleListenerIT > testListenerFiresOnIndexCreate PASSED

And, when run against a live node, the plugin log on index creation:

[INFO ][o.e.s.IndexLifecycleLogger] index created: [watch-me] (cluster state v37)

Round-trip test:

WatchMetadataTests > testSerialization PASSED

Stretch Goals

1. Mutate the custom metadata via an update task. Add a ClusterStateUpdateTask (submitted from your listener-adjacent code on the manager) that increments WatchMetadata.createdCount whenever an index is created, building a new Metadata with ClusterState.builder(currentState). This crosses you from observing state to changing it — and forces you to respect immutability.
2. Expose it via REST. Add a GET /_watch/stats handler (Lab 3.3 pattern) that reads clusterService.state().metadata().custom(WatchMetadata.TYPE) and returns created_count.
3. Multi-node integration test. Convert to OpenSearchIntegTestCase with @ClusterScope(numDataNodes = 2) and assert the custom metadata is identical on both nodes after a create — proving it published correctly. (Full treatment in Level 5.)

Validation / Self-check

1. Your listener runs on the applier thread. Name one thing it must never do, and explain the cluster-wide consequence if it does.
2. Why does a custom Metadata.Custom need entries in both NamedWriteableRegistry and NamedXContentRegistry? What fails if you register only one?
3. ClusterChangedEvent.indicesCreated() already gives you the delta. What did the framework diff to produce it, and on which two states?
4. You wrote custom metadata on the manager and it appeared on the manager but broke applies on the other nodes. What is the single most likely missing line, and why does it only fail on the receiving nodes?
5. When would you implement a ClusterStateApplier instead of a ClusterStateListener for this plugin? Give a concrete example.
6. Your round-trip test serializes a random WatchMetadata and asserts equality. Which production failure mode does that test specifically prevent?

When the listener test passes (and, ideally, the custom-metadata round-trip too), you've completed Lab 4.3. Continue to Lab 4.4: Fix It — An AllocationDecider Edge Case.