java - 点燃连续查询: Failed to initialize a remote filter

Question

我有 1 个服务器节点和 2 个客户端节点。第一个客户端将条目放入缓存(my-cache)，并且另一种使用连续查询来执行某些操作。下面是连续查询的代码片段:

IgniteCache<Integer, MyDomain> cache = ignite.cache("my-cache");
ContinuousQuery<Integer, MyDomain> qry = new ContinuousQuery<>();

qry.setLocalListener(evts -> {
    for (CacheEntryEvent<? extends Integer, ? extends MyDomain> e : evts) {
        doSomething();
    }
});

qry.setRemoteFilter(evt -> doSomething(evt));
cache.query(qry);

当我尝试启动集群时，我在第二个客户端节点上遇到以下异常:

    [19:03:36] (err) Failed to notify listener: o.a.i.i.util.future.GridFutureChainListener@2e28f5a0class org.apache.ignite.IgniteException: Failed to initialize a remote filter.
        at org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler.lambda$register$2bf956f5$1(CacheContinuousQueryHandler.java:337)
        at org.apache.ignite.internal.util.future.GridFutureChainListener.applyCallback(GridFutureChainListener.java:78)
        at org.apache.ignite.internal.util.future.GridFutureChainListener.apply(GridFutureChainListener.java:70)
        at org.apache.ignite.internal.util.future.GridFutureChainListener.apply(GridFutureChainListener.java:30)
        at org.apache.ignite.internal.util.future.GridFutureAdapter.notifyListener(GridFutureAdapter.java:385)
        at org.apache.ignite.internal.util.future.GridFutureAdapter.unblock(GridFutureAdapter.java:349)
        at org.apache.ignite.internal.util.future.GridFutureAdapter.unblockAll(GridFutureAdapter.java:337)
        at org.apache.ignite.internal.util.future.GridFutureAdapter.onDone(GridFutureAdapter.java:497)
        at org.apache.ignite.internal.util.future.GridFutureAdapter.onDone(GridFutureAdapter.java:476)
        at org.apache.ignite.internal.util.future.GridFutureAdapter.onDone(GridFutureAdapter.java:464)
        at org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler.p2pUnmarshal(CacheContinuousQueryHandler.java:1231)
        at org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler.p2pUnmarshal(CacheContinuousQueryHandler.java:1203)
        at org.apache.ignite.internal.processors.continuous.GridContinuousProcessor$8.run(GridContinuousProcessor.java:689)
        at org.apache.ignite.internal.util.IgniteUtils.wrapThreadLoader(IgniteUtils.java:6864)
        at org.apache.ignite.internal.processors.closure.GridClosureProcessor$1.body(GridClosureProcessor.java:827)
        at org.apache.ignite.internal.util.worker.GridWorker.run(GridWorker.java:120)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: class org.apache.ignite.internal.IgniteDeploymentCheckedException: Failed to obtain deployment for class: com.abc.service.MyService$$Lambda$286/1896828359
        at org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryDeployableObject.unmarshal(CacheContinuousQueryDeployableObject.java:95)
        at org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler.p2pUnmarshal(CacheContinuousQueryHandler.java:1228)
        ... 8 more

Answer 1

我为解决该问题所做的工作如下，

创建了两个类:

public class CacheEntryEventFilterImpl implements CacheEntryEventFilter<Integer, MyDomain> {
    @Override
    public boolean evaluate(CacheEntryEvent<? extends Integer, ? extends MyDomain> event)
            throws CacheEntryListenerException {
        return doSomething();
    }
}


public class FactoryImpl implements Factory<CacheEntryEventFilter<Integer, MyDomain>> {
    private static final long serialVersionUID = 881669537085864136L;

    @Override
    public CacheEntryEventFilter<Integer, MyDomain> create() {
        return new CacheEntryEventFilterImpl();
    }
}

在每个节点上都可以使用上述类。最后将 qry.setRemoteFilter(evt -> doSomething(evt)); 替换为 qry.setRemoteFilterFactory(new FactoryImpl());