@Test
public void testBuildWithPredicate() {

    CacheConfiguration<Integer, Integer> upstreamCacheConfiguration = new CacheConfiguration<>();
    upstreamCacheConfiguration.setAffinity(new RendezvousAffinityFunction(false, 1));
    upstreamCacheConfiguration.setName(UUID.randomUUID().toString());
    IgniteCache<Integer, Integer> upstreamCache = ignite.createCache(upstreamCacheConfiguration);
    upstreamCache.put(1, 1);
    upstreamCache.put(2, 2);
    CacheBasedDatasetBuilder<Integer, Integer> builder = new CacheBasedDatasetBuilder<>(
        ignite,
        upstreamCache,
        (k, v) -> k % 2 == 0
    );
    CacheBasedDataset<Integer, Integer, Long, AutoCloseable> dataset = builder.build(
        TestUtils.testEnvBuilder(),
        (env, upstream, upstreamSize) -> {
            UpstreamEntry<Integer, Integer> entry = upstream.next();
            assertEquals(Integer.valueOf(2), entry.getKey());
            assertEquals(Integer.valueOf(2), entry.getValue());
            assertFalse(upstream.hasNext());
            return 0L;
        },
        (env, upstream, upstreamSize, ctx) -> {
            UpstreamEntry<Integer, Integer> entry = upstream.next();
            assertEquals(Integer.valueOf(2), entry.getKey());
            assertEquals(Integer.valueOf(2), entry.getValue());
            assertFalse(upstream.hasNext());
            return null;
        },
        TestUtils.testEnvBuilder().buildForTrainer()
    );
    dataset.compute(data -> {});
}