How to create timer task (cron-based) without it execution on each node i.e. one execution per one time point in timetable using Apache Ingite?
I have cluster that consists in 2 nodes and application (war) with timer-task. In non-cluster mode application works well. But it has inside timer-task (ie. start each 5 minutes), that works with shared resources.
I try to do it. But IngiteScheduler#scheduleLocal deploys and runs task on each node, if both application instance started (each application instance try to start same timer task).
I assume that ignite have mechanism for deploy task with id fot it...
Thanks.
(thank @alamar for idea)
Below I show source and test for successful solutuion:
IgniteTimerTest.java:
package com.stackoverflow.question53780890.test;
import com.stackoverflow.question53780890.JobRunner;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteServices;
import org.apache.ignite.resources.SpringResource;
import org.apache.ignite.services.Service;
import org.apache.ignite.services.ServiceContext;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.ImportResource;
import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
import org.springframework.core.io.ClassPathResource;
import java.io.Serializable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class IgniteTimerTest {
private static final ExecutorService ES = Executors.newFixedThreadPool(5);
@Test
public void test() throws Exception {
Future<ConfigurableApplicationContext> applicationContextFutureOne = ES.submit(() -> create(ConfigOne.class));
Future<ConfigurableApplicationContext> applicationContextFutureTwo = ES.submit(() -> create(ConfigTwo.class));
try (ConfigurableApplicationContext applicationContextOne = applicationContextFutureOne.get();
ConfigurableApplicationContext applicationContextTwo = applicationContextFutureTwo.get();) {
Ignite igniteOne = applicationContextOne.getBean(Ignite.class);
Ignite igniteTwo = applicationContextTwo.getBean(Ignite.class);
IgniteServices servicesOne = igniteOne.services();
IgniteServices servicesTwo = igniteTwo.services();
servicesOne.deployClusterSingleton(TestTimerRunner.class.getName(), new TestTimerRunner());
Thread.sleep(JobRunner.PERIOD * 3);
servicesTwo.deployClusterSingleton(TestTimerRunner.class.getName(), new TestTimerRunner());
Thread.sleep(JobRunner.PERIOD * 3);
applicationContextOne.close();
Thread.sleep(JobRunner.PERIOD * 3);
int countValue = JobRunner.getConterValue();
Assertions.assertTrue(9 <= countValue && countValue <= 11);
}
}
private ConfigurableApplicationContext create(Class mainClass) {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(mainClass);
return context;
}
class TestTimerRunner implements Serializable, Service {
@SpringResource(resourceClass = JobRunner.class)
private transient JobRunner jobRunner;
public TestTimerRunner() {
}
@Override
public void cancel(ServiceContext ctx) {
jobRunner.stop();
}
@Override
public void init(ServiceContext ctx) throws Exception {
}
@Override
public void execute(ServiceContext ctx) throws Exception {
jobRunner.start();
}
}
@Configuration
@ImportResource("classpath:common.xml")
public static class ConfigOne {
@Bean
public static PropertySourcesPlaceholderConfigurer placeHolderConfigurer() {
PropertySourcesPlaceholderConfigurer holder = new PropertySourcesPlaceholderConfigurer();
holder.setLocation(new ClassPathResource("one.properties"));
return holder;
}
}
@Configuration
@ImportResource("classpath:common.xml")
public static class ConfigTwo {
@Bean
public static PropertySourcesPlaceholderConfigurer placeHolderConfigurer() {
PropertySourcesPlaceholderConfigurer holder = new PropertySourcesPlaceholderConfigurer();
holder.setLocation(new ClassPathResource("two.properties"));
return holder;
}
}
}
JobRunner.java:
package com.stackoverflow.question53780890;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.TaskScheduler;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicInteger;
public class JobRunner implements Runnable {
public static final long PERIOD = 5_000;
private static AtomicInteger counter = new AtomicInteger();
@Autowired
private TaskScheduler taskScheduler;
private ScheduledFuture<?> job;
public static int getConterValue() {
return counter.get();
}
public void start() {
job = taskScheduler.scheduleAtFixedRate(this, PERIOD);
}
public void stop() {
job.cancel(true);
}
@Override
public void run() {
counter.incrementAndGet();
}
}
common.xml:
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:c="http://www.springframework.org/schema/c"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">
<task:annotation-driven executor="myExecutor" scheduler="myScheduler"/>
<task:executor id="myExecutor" pool-size="5"/>
<task:scheduler id="myScheduler" pool-size="5"/>
<bean id="jobRunner" class="com.stackoverflow.question53780890.JobRunner"/>
<bean id="ignite" class="org.apache.ignite.IgniteSpringBean">
<property name="configuration" ref="igniteConfiguration"/>
</bean>
<bean id="igniteConfiguration" class="org.apache.ignite.configuration.IgniteConfiguration">
<property name="igniteInstanceName" value="${igniteInstanceName}"/>
<property name="discoverySpi">
<bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
<property name="ipFinder">
<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder"
p:addresses-ref="igniteNodesAddresses"/>
</property>
<property name="joinTimeout" value="20000"/>
<property name="localPort" value="${igniteLocalPort}"/>
<property name="localPortRange" value="1"/>
</bean>
</property>
<property name="communicationSpi">
<bean class="org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi">
<property name="localPort" value="${communicationSpiPort}"/>
</bean>
</property>
</bean>
<bean id="igniteNodesAddresses" class="org.springframework.util.StringUtils"
factory-method="commaDelimitedListToStringArray">
<constructor-arg type="java.lang.String" value="${igniteNodes}"/>
</bean>
</beans>
one.properties:
igniteLocalPort=47501
communicationSpiPort=47101
igniteNodes=localhost:47501,localhost:47502
igniteInstanceName=one
two.properties:
igniteLocalPort=47502
communicationSpiPort=47102
igniteNodes=localhost:47501,localhost:47502
igniteInstanceName=two
I think you could deploy a singleton service which will do scheduleLocal in its execute()
and de-schedule in its cancel()
method. This way it will be failed over to next node and properly re-schedule your tasks if current node leaves.