class CacheBasedDatasetBuilder implements DatasetBuilder<K, V> {

    @SuppressWarnings("unchecked")
    @Override public <C extends Serializable, D extends AutoCloseable> CacheBasedDataset<K, V, C, D> build(
        LearningEnvironmentBuilder envBuilder,
        PartitionContextBuilder<K, V, C> partCtxBuilder,
        PartitionDataBuilder<K, V, C, D> partDataBuilder,
        LearningEnvironment localLearningEnv) {

        UUID datasetId = UUID.randomUUID();

        // Retrieves affinity function of the upstream Ignite Cache.
        CacheConfiguration<K, V> upstreamCacheConfiguration = upstreamCache.getConfiguration(CacheConfiguration.class);
        AffinityFunction upstreamCacheAffinity = upstreamCacheConfiguration.getAffinity();

        // Creates dataset cache configuration with affinity function that mimics to affinity function of the upstream
        // cache.
        CacheConfiguration<Integer, C> datasetCacheConfiguration = new CacheConfiguration<>();
        datasetCacheConfiguration.setName(String.format(DATASET_CACHE_TEMPLATE, upstreamCache.getName(), datasetId));
        datasetCacheConfiguration.setAffinity(new DatasetAffinityFunctionWrapper(upstreamCacheAffinity));

        IgniteCache<Integer, C> datasetCache = ignite.createCache(datasetCacheConfiguration);

        ComputeUtils.initContext(
            ignite,
            upstreamCache.getName(),
            transformerBuilder,
            filter,
            datasetCache.getName(),
            partCtxBuilder,
            envBuilder,
            retries,
            RETRY_INTERVAL,
            upstreamKeepBinary,
            localLearningEnv.deployingContext()
        );

        return new CacheBasedDataset<>(ignite, upstreamCache, filter, transformerBuilder, datasetCache, envBuilder, partDataBuilder, datasetId, upstreamKeepBinary, localLearningEnv, retries);
    }

    public  CacheBasedDatasetBuilder(Ignite ignite, IgniteCache<K, V> upstreamCache);
    public  CacheBasedDatasetBuilder(Ignite ignite, IgniteCache<K, V> upstreamCache, IgniteBiPredicate<K, V> filter);
    public  CacheBasedDatasetBuilder(Ignite ignite,
        IgniteCache<K, V> upstreamCache,
        IgniteBiPredicate<K, V> filter,
        UpstreamTransformerBuilder transformerBuilder);
    public  CacheBasedDatasetBuilder(Ignite ignite,
        IgniteCache<K, V> upstreamCache,
        IgniteBiPredicate<K, V> filter,
        UpstreamTransformerBuilder transformerBuilder,
        Boolean isKeepBinary,
        int retries);

}