Romeh

    @Test
    public void testMapReducedJobsWithFailFastSync(){
        // example of ignite jobs, first one succeeded , second fail, third succeeded , but the reducer will fail fast once he collect the failed job
        IgniteCallable validationServiceJob1=() -> ServiceResponse.<String>builder().response("Job 1 is valid").serviceOrigin("job1")
                .success(true).build();
        IgniteCallable validationServiceJob2=() -> ServiceResponse.<String>builder().response("Job 2 is failed").serviceOrigin("job2")
                .success(false).build();
        IgniteCallable validationServiceJob3=() -> ServiceResponse.<String>builder().response("Job 3 is valid").serviceOrigin("job3")
                .success(true).build();

Romeh

/**
 * sample service for how to call map reduce jobs in parallel asynchronous with fail fast reducer
 */
@Service
public class ComputeService {

    private static final Logger logger = LoggerFactory.getLogger(AlertsService.class);
    private final DataGridCompute dataGridCompute;
    @Autowired
    private  FailFastReducer failFastReducer;

Romeh

/**
 * @param <T> the service call response type
 */
@Getter
@Setter
@ToString
@EqualsAndHashCode
@Builder
public class ServiceResponse<T> implements Serializable {
    private T response;

Romeh

/**
 *  the generic reduce response that contain all single collected jobs responses
 */
@Builder
@Getter
@ToString
@EqualsAndHashCode
public class MapReduceResponse implements Serializable {
    private Map<String, ServiceResponse> reducedResponses;
    boolean success;

Romeh

/**
 * generic utility class for map reduce call
 */
@Component
public class DataGridCompute {

    @Autowired
    private Ignite ignite;

    /**

Romeh

/**
 * a fail fast map reducer to decide if it should keep waiting for other jobs to final reduce or it should terminate
 * and fail fast with the current responses if any failed
 */
@Component
@Scope("prototype")
public class FailFastReducer implements IgniteReducer<ServiceResponse, MapReduceResponse> {

    private final  Map<String, ServiceResponse> responseMap = new ConcurrentHashMap<>();

Romeh

/**
 * the main ignite snapshot implementation based into SnapshotStore
 */
@Slf4j
public class IgniteSnapshotStore extends SnapshotStore {

    private final Serializer serializer;
    private final Store<SnapshotItem> storage;
    private final IgniteCache<Long, SnapshotItem> cache;
    private final BiFunction<Config, ActorSystem, IgniteCache<Long, SnapshotItem>> snapshotCacheProvider =

Romeh

/**
 * the main ignite journal plugin implementation based into AsyncWriteJournal
 */
@Slf4j
public class IgniteWriteJournal extends AsyncWriteJournal {

    private final Serializer serializer;
    private final Store<JournalItem> storage;
    private final IgniteCache<Long, JournalItem> cache;
    private final IgniteCache<String, Long> sequenceNumberTrack;

Romeh

  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::        (v1.5.6.RELEASE)

[2017-12-03 19:48:24] [main] INFO  o.s.c.a.AnnotationConfigApplicationContext - Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@2c2db130: startup date [Sun Dec 03 19:48:24 CET 2017]; root of context hierarchy
[2017-12-03 19:48:24] [main] INFO  o.s.c.c.s.e.NativeEnvironmentRepository - Adding property source: file:/var/folders/dl/660f5pxs0q18dmp4zh0smfgh0000gn/T/config-repo-5816360046828309118/Application.yml

Romeh

spring:
# change the application name for your project name , this name is reserved for the maven archetype code generation
  application:
    name: Application

---
# DO NOT FORGET TO ADD YOUR YAML CONFIG FILE IN config server as shown below
spring:
  cloud:
    config: