alibaba/canal · error · IllegalArgumentException

CanalEventSink must be EntryEventSink

Error message

CanalEventSink must be EntryEventSink

What it means

EntryCollector extracts per-entry metrics (execute time, transaction count) by attaching a PrometheusCanalEventDownStreamHandler to the instance's CanalEventSink. This only works if the sink is an EntryEventSink that exposes the handler list, so register() throws IllegalArgumentException otherwise. The instance must be wired with an EntryEventSink.

Source

Thrown at prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/EntryCollector.java:73

            long latest = emh.latestExecTime.get();
            // execTime > now，delay显示为0
            long d = (now >= latest) ? (now - latest) : 0;
            delay.addMetric(emh.destLabelValues, d);
            transactions.addMetric(emh.destLabelValues, emh.transactionCounter.doubleValue());
        }
        mfs.add(delay);
        mfs.add(transactions);
        return mfs;
    }

    @Override
    public void register(CanalInstance instance) {
        final String destination = instance.getDestination();
        EntryMetricsHolder holder = new EntryMetricsHolder();
        holder.destLabelValues = Collections.singletonList(destination);
        CanalEventSink sink = instance.getEventSink();
        if (!(sink instanceof EntryEventSink)) {
            throw new IllegalArgumentException("CanalEventSink must be EntryEventSink");
        }
        EntryEventSink entrySink = (EntryEventSink) sink;
        PrometheusCanalEventDownStreamHandler handler = assembleHandler(entrySink);
        holder.latestExecTime = handler.getLatestExecuteTime();
        holder.transactionCounter = handler.getTransactionCounter();
        Preconditions.checkNotNull(holder.latestExecTime);
        Preconditions.checkNotNull(holder.transactionCounter);
        EntryMetricsHolder old = instances.put(destination, holder);
        if (old != null) {
            logger.warn("Remove stale EntryCollector for instance {}.", destination);
        }
    }

    @Override
    public void unregister(CanalInstance instance) {
        final String destination = instance.getDestination();
        CanalEventSink sink = instance.getEventSink();
        if (!(sink instanceof EntryEventSink)) {

View on GitHub (pinned to 87be50e876)

Solutions

1. Configure the canal instance to use com.alibaba.otter.canal.sink.entry.EntryEventSink as its event sink.
2. If prometheus metrics are not needed, exclude the EntryCollector / disable prometheus to avoid the type check.
3. For custom instances, ensure getEventSink() returns an EntryEventSink-compatible sink before metrics registration.

Example fix

// before
instance.setEventSink(new MyCustomSink()); // not EntryEventSink
entryCollector.register(instance);

// after
EntryEventSink sink = new EntryEventSink();
instance.setEventSink(sink);
entryCollector.register(instance);

CanalEventSink sink = instance.getEventSink();
if (!(sink instanceof EntryEventSink)) {
    throw new IllegalStateException("prometheus EntryCollector requires EntryEventSink, got " + sink.getClass());
}
entryCollector.register(instance);

boolean isEntryEventSink(CanalEventSink s) { return s instanceof EntryEventSink; }

When it happens

Trigger: Calling EntryCollector.register(instance) when instance.getEventSink() returns a CanalEventSink that is not an EntryEventSink (e.g. a custom or no-op sink implementation).

Common situations: A custom canal instance built without EntryEventSink; a deployer version where the default sink class changed; a test instance using a mock/stub sink.

Related errors

• Canal prometheus collector need to implement InstanceRegistr
• No kudu adapter found for config key: {}
• not allow to change outAdapterKey
• KuduMapping.database
• ERROR load Config: {} {}

AI-assisted analysis of alibaba/canal@87be50e876 (2026-08-14). Data as JSON: /api/errors/1b9559f4a7c2c6dc. Report an issue: GitHub.