Apache Ignite使用NodeFilter实现多节点时分区与数据节点异常问题咨询

问题背景 #

尝试寻找简单的Apache Ignite多节点部署方案，因标准affinity配置复杂，计划通过缓存的nodeFilter将特定缓存绑定到指定节点，实现同组缓存集中部署，后续按分区发送任务。但启用nodeFilter后出现异常：预期缓存数据会分布在目标节点的本地分区中，实际affinity返回0个分区，forDataNodes()方法返回的节点数也为0。

问题原因解释 #

核心错误出在nodeFilter的匹配逻辑上：
代码中尝试用node.consistentId().equals(nodes.get(0))做过滤条件，但nodes.get(0)是ClusterNode对象，而node.consistentId()返回的是字符串类型，两者类型不匹配，导致过滤条件永远无法成立。最终没有节点满足缓存部署要求，缓存无法分配任何数据节点，因此出现分区数为0、数据节点数为0的现象。

解决方向 #

1. 修正nodeFilter匹配逻辑
   将过滤条件改为直接比较节点的consistentId字符串，确保能正确匹配目标节点：

   // 正确的过滤逻辑：比较consistentId字符串
   ccfg.setNodeFilter(node -> CONSISTENT_ID.equals(node.consistentId()));
   

2. 提前验证目标节点存在性
   在设置nodeFilter前，先确认筛选到的目标节点不为空，避免因节点不存在导致缓存无数据节点的情况：

   List<ClusterNode> nodes = ignite.cluster().nodes().stream()
           .filter(n -> CONSISTENT_ID.equals(n.consistentId()))
           .collect(Collectors.toList());
   
   if (nodes.isEmpty()) {
       throw new IllegalStateException("未找到目标节点: " + CONSISTENT_ID);
   }
   

3. 保持同组缓存配置一致性
   若要确保同组缓存分配到同一节点的相同分区，需保证这些缓存使用相同的groupName和AffinityFunction配置（当前代码已设置相同groupName，此部分无需修改）。

修正后的测试代码示例 #

public class TestIgnite {

    private static final String CONSISTENT_ID = "node1";
    private static Ignite ignite;
    
    @BeforeAll
    public static void prepareTest() throws InterruptedException {
        System.setProperty("IGNITE_OVERRIDE_CONSISTENT_ID", CONSISTENT_ID);
        Ignition.start(IgniteConfigurations.loadDefaultConfiguration());
        ignite = Ignition.ignite(ProcessingConfiguration.GRID_NAME);
        ignite.cluster().state(ClusterState.ACTIVE);
        Thread.sleep(5000);     
    }
    
    @Test
    public void testNodeFilter()  {
        String CACHE_NAME = "CACHE_NAME";
        CacheConfiguration<String, String> ccfg = new CacheConfiguration<>(CACHE_NAME);
        ccfg.setAtomicityMode(CacheAtomicityMode.ATOMIC);
        ccfg.setBackups(0);
        ccfg.setAffinity(new RendezvousAffinityFunction());
        ccfg.setDataRegionName("main");
        ccfg.setGroupName("main");
        ccfg.setSqlEscapeAll(true);
        ccfg.setCopyOnRead(false);
        ccfg.setEventsDisabled(true);
        ccfg.setCacheMode(CacheMode.PARTITIONED);       
        
        org.apache.ignite.configuration.NearCacheConfiguration<String, String> cfg = new org.apache.ignite.configuration.NearCacheConfiguration<>();
        cfg.setNearEvictionPolicyFactory(new LruEvictionPolicyFactory<>(100));
        ccfg.setNearConfiguration(cfg);
        
        // 修正后的节点筛选与过滤逻辑
        List<ClusterNode> nodes = ignite.cluster().nodes().stream()
                .filter(n -> CONSISTENT_ID.equals(n.consistentId()))
                .collect(Collectors.toList());
        
        if (nodes.isEmpty()) {
            throw new IllegalStateException("未找到目标节点: " + CONSISTENT_ID);
        }
        
        ccfg.setNodeFilter(node -> CONSISTENT_ID.equals(node.consistentId()));
            
        ignite.createCache(ccfg);
        
        ClusterGroup grpCache = ignite.cluster().forCacheNodes(CACHE_NAME);
        System.out.println("Cache group size : " + grpCache.nodes().size());
        
        ClusterGroup grpData = ignite.cluster().forDataNodes(CACHE_NAME);
        System.out.println("Data group size : " + grpData.nodes().size());

        int[] p = ignite.affinity(CACHE_NAME).allPartitions(grpCache.node());
        System.out.println("# of partitions : " + p.length);
        
        assertEquals(1024, p.length);
    }

    @AfterAll
    public static void tearDown() {
        ignite.close();
    }
}

内容的提问来源于stack exchange，提问作者Steve Hostettler