# Renaming
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/725526fa Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/725526fa Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/725526fa Branch: refs/heads/master Commit: 725526fa165fa3137a6e72cdcae6b96ce0e79a2f Parents: 50d1554 Author: sboikov <sboi...@gridgain.com> Authored: Fri Dec 5 17:00:01 2014 +0300 Committer: sboikov <sboi...@gridgain.com> Committed: Fri Dec 5 17:00:04 2014 +0300 ---------------------------------------------------------------------- .../java/org/apache/ignite/IgniteCompute.java | 2 +- .../compute/ComputeJobBeforeFailover.java | 2 +- .../ignite/compute/ComputeJobContext.java | 2 +- .../org/apache/ignite/compute/ComputeTask.java | 2 +- .../apache/ignite/compute/ComputeTaskSpis.java | 2 +- .../configuration/IgniteConfiguration.java | 16 +- .../jobstealing/JobStealingCollisionSpi.java | 2 +- .../spi/eventstorage/EventStorageSpi.java | 57 +++ .../memory/MemoryEventStorageSpi.java | 272 +++++++++++++++ .../memory/MemoryEventStorageSpiMBean.java | 57 +++ .../ignite/spi/eventstorage/memory/package.html | 15 + .../apache/ignite/spi/eventstorage/package.html | 15 + .../ignite/spi/failover/FailoverContext.java | 47 +++ .../apache/ignite/spi/failover/FailoverSpi.java | 60 ++++ .../spi/failover/always/AlwaysFailoverSpi.java | 238 +++++++++++++ .../failover/always/AlwaysFailoverSpiMBean.java | 36 ++ .../ignite/spi/failover/always/package.html | 15 + .../jobstealing/JobStealingFailoverSpi.java | 343 +++++++++++++++++++ .../JobStealingFailoverSpiMBean.java | 46 +++ .../spi/failover/jobstealing/package.html | 15 + .../spi/failover/never/NeverFailoverSpi.java | 94 +++++ .../failover/never/NeverFailoverSpiMBean.java | 21 ++ .../ignite/spi/failover/never/package.html | 15 + .../org/apache/ignite/spi/failover/package.html | 15 + .../org/gridgain/grid/kernal/GridGainEx.java | 8 +- .../eventstorage/GridEventStorageManager.java | 2 +- .../failover/GridFailoverContextImpl.java | 2 +- .../managers/failover/GridFailoverManager.java | 2 +- .../grid/spi/eventstorage/EventStorageSpi.java | 57 --- .../memory/MemoryEventStorageSpi.java | 272 --------------- .../memory/MemoryEventStorageSpiMBean.java | 57 --- .../grid/spi/eventstorage/memory/package.html | 15 - .../gridgain/grid/spi/eventstorage/package.html | 15 - .../grid/spi/failover/FailoverContext.java | 47 --- .../gridgain/grid/spi/failover/FailoverSpi.java | 60 ---- .../spi/failover/always/AlwaysFailoverSpi.java | 238 ------------- .../failover/always/AlwaysFailoverSpiMBean.java | 36 -- .../grid/spi/failover/always/package.html | 15 - .../jobstealing/JobStealingFailoverSpi.java | 343 ------------------- .../JobStealingFailoverSpiMBean.java | 46 --- .../grid/spi/failover/jobstealing/package.html | 15 - .../spi/failover/never/NeverFailoverSpi.java | 94 ----- .../failover/never/NeverFailoverSpiMBean.java | 21 -- .../grid/spi/failover/never/package.html | 15 - .../org/gridgain/grid/spi/failover/package.html | 15 - .../src/test/config/job-loadtest/client.xml | 2 +- .../src/test/config/job-loadtest/server.xml | 2 +- .../core/src/test/config/load/dsi-load-base.xml | 2 +- ...bStealingCollisionSpiAttributesSelfTest.java | 2 +- ...alingCollisionSpiCustomTopologySelfTest.java | 2 +- .../GridJobStealingCollisionSpiSelfTest.java | 2 +- ...obStealingCollisionSpiStartStopSelfTest.java | 2 +- ...MemoryEventStorageMultiThreadedSelfTest.java | 43 +++ ...GridMemoryEventStorageSpiConfigSelfTest.java | 26 ++ .../GridMemoryEventStorageSpiSelfTest.java | 130 +++++++ ...dMemoryEventStorageSpiStartStopSelfTest.java | 21 ++ .../ignite/spi/eventstorage/memory/package.html | 15 + .../spi/failover/GridFailoverTestContext.java | 62 ++++ .../GridAlwaysFailoverSpiConfigSelfTest.java | 25 ++ .../always/GridAlwaysFailoverSpiSelfTest.java | 110 ++++++ .../GridAlwaysFailoverSpiStartStopSelfTest.java | 22 ++ .../ignite/spi/failover/always/package.html | 15 + ...ridJobStealingFailoverSpiConfigSelfTest.java | 25 ++ ...idJobStealingFailoverSpiOneNodeSelfTest.java | 88 +++++ .../GridJobStealingFailoverSpiSelfTest.java | 217 ++++++++++++ ...JobStealingFailoverSpiStartStopSelfTest.java | 46 +++ .../spi/failover/jobstealing/package.html | 15 + .../never/GridNeverFailoverSpiSelfTest.java | 37 ++ .../GridNeverFailoverSpiStartStopSelfTest.java | 22 ++ .../ignite/spi/failover/never/package.html | 15 + .../org/apache/ignite/spi/failover/package.html | 15 + .../GridAlwaysFailoverSpiFailSelfTest.java | 4 +- .../GridFailoverCustomTopologySelfTest.java | 4 +- .../grid/kernal/GridFailoverSelfTest.java | 2 +- .../GridFailoverTaskWithPredicateSelfTest.java | 4 +- .../kernal/GridFailoverTopologySelfTest.java | 4 +- .../grid/kernal/GridJobStealingSelfTest.java | 2 +- .../GridJobStealingZeroActiveJobsSelfTest.java | 2 +- .../grid/kernal/GridMultipleSpisSelfTest.java | 4 +- .../GridMultithreadedJobStealingSelfTest.java | 2 +- .../grid/kernal/GridSpiExceptionSelfTest.java | 2 +- .../grid/kernal/GridStopWithWaitSelfTest.java | 2 +- .../managers/GridManagerStopSelfTest.java | 6 +- .../GridCacheGroupLockFailoverSelfTest.java | 4 +- .../cache/GridCachePutAllFailoverSelfTest.java | 4 +- .../GridCachePreloadEventsAbstractSelfTest.java | 2 +- .../GridCacheReplicatedPreloadSelfTest.java | 2 +- .../p2p/GridAbstractMultinodeRedeployTest.java | 2 +- .../session/GridSessionJobFailoverSelfTest.java | 2 +- ...MemoryEventStorageMultiThreadedSelfTest.java | 43 --- ...GridMemoryEventStorageSpiConfigSelfTest.java | 26 -- .../GridMemoryEventStorageSpiSelfTest.java | 130 ------- ...dMemoryEventStorageSpiStartStopSelfTest.java | 21 -- .../grid/spi/eventstorage/memory/package.html | 15 - .../spi/failover/GridFailoverTestContext.java | 62 ---- .../GridAlwaysFailoverSpiConfigSelfTest.java | 25 -- .../always/GridAlwaysFailoverSpiSelfTest.java | 110 ------ .../GridAlwaysFailoverSpiStartStopSelfTest.java | 22 -- .../grid/spi/failover/always/package.html | 15 - ...ridJobStealingFailoverSpiConfigSelfTest.java | 25 -- ...idJobStealingFailoverSpiOneNodeSelfTest.java | 88 ----- .../GridJobStealingFailoverSpiSelfTest.java | 217 ------------ ...JobStealingFailoverSpiStartStopSelfTest.java | 46 --- .../grid/spi/failover/jobstealing/package.html | 15 - .../never/GridNeverFailoverSpiSelfTest.java | 37 -- .../GridNeverFailoverSpiStartStopSelfTest.java | 22 -- .../grid/spi/failover/never/package.html | 15 - .../org/gridgain/grid/spi/failover/package.html | 15 - .../direct/stealing/GridStealingLoadTest.java | 2 +- .../GridSpiEventStorageSelfTestSuite.java | 2 +- .../GridSpiFailoverSelfTestSuite.java | 6 +- pom.xml | 8 +- 112 files changed, 2374 insertions(+), 2374 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/IgniteCompute.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteCompute.java b/modules/core/src/main/java/org/apache/ignite/IgniteCompute.java index 309a547..4a8cd49 100644 --- a/modules/core/src/main/java/org/apache/ignite/IgniteCompute.java +++ b/modules/core/src/main/java/org/apache/ignite/IgniteCompute.java @@ -63,7 +63,7 @@ import java.util.concurrent.*; * or has rejected execution due to lack of resources. By default, in case of failover, next * load balanced node will be picked for job execution. Also jobs will never be re-routed to the * nodes they have failed on. This behavior can be changed by configuring any of the existing or a custom - * {@link org.gridgain.grid.spi.failover.FailoverSpi} in grid configuration. + * {@link org.apache.ignite.spi.failover.FailoverSpi} in grid configuration. * <h1 class="header">Resource Injection</h1> * All compute jobs, including closures, runnables, callables, and tasks can be injected with * grid resources. Both, field and method based injections are supported. The following grid http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobBeforeFailover.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobBeforeFailover.java b/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobBeforeFailover.java index 48067c6..82002eb 100644 --- a/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobBeforeFailover.java +++ b/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobBeforeFailover.java @@ -13,7 +13,7 @@ import java.lang.annotation.*; /** * This annotation allows to call a method right before job is submitted to - * {@link org.gridgain.grid.spi.failover.FailoverSpi}. In this method job can re-create necessary state that was + * {@link org.apache.ignite.spi.failover.FailoverSpi}. In this method job can re-create necessary state that was * cleared, for example, in method with {@link ComputeJobAfterSend} annotation. * <p> * This annotation can be applied to methods of {@link ComputeJob} instances only. It is http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobContext.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobContext.java b/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobContext.java index a03641d..47174ad 100644 --- a/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobContext.java +++ b/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobContext.java @@ -35,7 +35,7 @@ import java.util.concurrent.*; * For example, if you need to cancel an actively running job from {@link org.apache.ignite.spi.collision.CollisionSpi} * you may choose to set some context attribute on the job to mark the fact * that a job was cancelled by grid and not by a user. Context attributes can - * also be assigned in {@link org.gridgain.grid.spi.failover.FailoverSpi} prior to failing over a job. + * also be assigned in {@link org.apache.ignite.spi.failover.FailoverSpi} prior to failing over a job. * <p> * From within {@link ComputeTask#result(ComputeJobResult, List)} or {@link ComputeTask#reduce(List)} methods, * job context is available via {@link ComputeJobResult#getJobContext()} method which gives user the http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/compute/ComputeTask.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/compute/ComputeTask.java b/modules/core/src/main/java/org/apache/ignite/compute/ComputeTask.java index cb81e8c..2340def 100644 --- a/modules/core/src/main/java/org/apache/ignite/compute/ComputeTask.java +++ b/modules/core/src/main/java/org/apache/ignite/compute/ComputeTask.java @@ -72,7 +72,7 @@ import java.util.*; * <li> * If {@link ComputeJobResultPolicy#FAILOVER} policy is returned, then job will * be failed over to another node for execution. The node to which job will get - * failed over is decided by {@link org.gridgain.grid.spi.failover.FailoverSpi} SPI implementation. + * failed over is decided by {@link org.apache.ignite.spi.failover.FailoverSpi} SPI implementation. * Note that if you use {@link ComputeTaskAdapter} adapter for {@code GridComputeTask} * implementation, then it will automatically fail jobs to another node for 2 * known failure cases: http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/compute/ComputeTaskSpis.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/compute/ComputeTaskSpis.java b/modules/core/src/main/java/org/apache/ignite/compute/ComputeTaskSpis.java index 1598d47..53941f0 100644 --- a/modules/core/src/main/java/org/apache/ignite/compute/ComputeTaskSpis.java +++ b/modules/core/src/main/java/org/apache/ignite/compute/ComputeTaskSpis.java @@ -17,7 +17,7 @@ import java.lang.annotation.*; * This annotation allows task to specify what SPIs it wants to use. * Starting with {@code GridGain 2.1} you can start multiple instances * of {@link GridLoadBalancingSpi}, - * {@link org.gridgain.grid.spi.failover.FailoverSpi}, and {@link org.apache.ignite.spi.checkpoint.CheckpointSpi}. If you do that, + * {@link org.apache.ignite.spi.failover.FailoverSpi}, and {@link org.apache.ignite.spi.checkpoint.CheckpointSpi}. If you do that, * you need to tell a task which SPI to use (by default it will use the fist * SPI in the list). */ http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java b/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java index fc5349d..98f85a1 100644 --- a/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java +++ b/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java @@ -32,8 +32,8 @@ import org.apache.ignite.spi.collision.*; import org.apache.ignite.spi.communication.*; import org.apache.ignite.spi.deployment.*; import org.apache.ignite.spi.discovery.*; -import org.gridgain.grid.spi.eventstorage.*; -import org.gridgain.grid.spi.failover.*; +import org.apache.ignite.spi.eventstorage.*; +import org.apache.ignite.spi.failover.*; import org.gridgain.grid.spi.indexing.*; import org.gridgain.grid.spi.loadbalancing.*; import org.gridgain.grid.spi.loadbalancing.roundrobin.*; @@ -1752,7 +1752,7 @@ public class IgniteConfiguration { /** * Should return fully configured event SPI implementation. If not provided, - * {@link org.gridgain.grid.spi.eventstorage.memory.MemoryEventStorageSpi} will be used. + * {@link org.apache.ignite.spi.eventstorage.memory.MemoryEventStorageSpi} will be used. * * @return Grid event SPI implementation or {@code null} to use default implementation. */ @@ -1761,9 +1761,9 @@ public class IgniteConfiguration { } /** - * Sets fully configured instance of {@link org.gridgain.grid.spi.eventstorage.EventStorageSpi}. + * Sets fully configured instance of {@link org.apache.ignite.spi.eventstorage.EventStorageSpi}. * - * @param evtSpi Fully configured instance of {@link org.gridgain.grid.spi.eventstorage.EventStorageSpi}. + * @param evtSpi Fully configured instance of {@link org.apache.ignite.spi.eventstorage.EventStorageSpi}. * @see IgniteConfiguration#getEventStorageSpi() */ public void setEventStorageSpi(EventStorageSpi evtSpi) { @@ -2053,7 +2053,7 @@ public class IgniteConfiguration { /** * Should return fully configured failover SPI implementation. If not provided, - * {@link org.gridgain.grid.spi.failover.always.AlwaysFailoverSpi} will be used. + * {@link org.apache.ignite.spi.failover.always.AlwaysFailoverSpi} will be used. * * @return Grid failover SPI implementation or {@code null} to use default implementation. */ @@ -2062,9 +2062,9 @@ public class IgniteConfiguration { } /** - * Sets fully configured instance of {@link org.gridgain.grid.spi.failover.FailoverSpi}. + * Sets fully configured instance of {@link org.apache.ignite.spi.failover.FailoverSpi}. * - * @param failSpi Fully configured instance of {@link org.gridgain.grid.spi.failover.FailoverSpi} or + * @param failSpi Fully configured instance of {@link org.apache.ignite.spi.failover.FailoverSpi} or * {@code null} if no SPI provided. * @see IgniteConfiguration#getFailoverSpi() */ http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/collision/jobstealing/JobStealingCollisionSpi.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/collision/jobstealing/JobStealingCollisionSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/collision/jobstealing/JobStealingCollisionSpi.java index 0e356af..dd3c145 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/collision/jobstealing/JobStealingCollisionSpi.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/collision/jobstealing/JobStealingCollisionSpi.java @@ -54,7 +54,7 @@ import static org.apache.ignite.events.IgniteEventType.*; * <p> * <i> * Note that this SPI must always be used in conjunction with - * {@link org.gridgain.grid.spi.failover.jobstealing.JobStealingFailoverSpi}. + * {@link org.apache.ignite.spi.failover.jobstealing.JobStealingFailoverSpi}. * Also note that job metrics update should be enabled in order for this SPI * to work properly (i.e. {@link org.apache.ignite.configuration.IgniteConfiguration#getMetricsUpdateFrequency()} * should be set to {@code 0} or greater value). http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/EventStorageSpi.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/EventStorageSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/EventStorageSpi.java new file mode 100644 index 0000000..4f57ba3 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/EventStorageSpi.java @@ -0,0 +1,57 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.eventstorage; + +import org.apache.ignite.events.*; +import org.apache.ignite.lang.*; +import org.apache.ignite.spi.*; + +import java.util.*; + +/** + * This SPI provides local node events storage. SPI allows for recording local + * node events and querying recorded local events. Every node during its life-cycle + * goes through a serious of events such as task deployment, task execution, job + * execution, etc. For + * performance reasons GridGain is designed to store all locally produced events + * locally. These events can be later retrieved using either distributed query: + * <ul> + * <li>{@link org.apache.ignite.IgniteEvents#remoteQuery(org.apache.ignite.lang.IgnitePredicate, long, int...)}</li> + * </ul> + * or local only query: + * <ul> + * <li>{@link org.apache.ignite.IgniteEvents#localQuery(org.apache.ignite.lang.IgnitePredicate, int...)}</li> + * </ul> + * <b>NOTE:</b> this SPI (i.e. methods in this interface) should never be used directly. SPIs provide + * internal view on the subsystem and is used internally by GridGain kernal. In rare use cases when + * access to a specific implementation of this SPI is required - an instance of this SPI can be obtained + * via {@link org.apache.ignite.Ignite#configuration()} method to check its configuration properties or call other non-SPI + * methods. Note again that calling methods from this interface on the obtained instance can lead + * to undefined behavior and explicitly not supported. + * @see org.apache.ignite.events.IgniteEvent + */ +public interface EventStorageSpi extends IgniteSpi { + /** + * Queries locally-stored events only. Events could be filtered out + * by given predicate filter. + * + * @param p Event predicate filter. + * @return Collection of events. + */ + public <T extends IgniteEvent> Collection<T> localEvents(IgnitePredicate<T> p); + + /** + * Records single event. + * + * @param evt Event that should be recorded. + * @throws org.apache.ignite.spi.IgniteSpiException If event recording failed for any reason. + */ + public void record(IgniteEvent evt) throws IgniteSpiException; +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/memory/MemoryEventStorageSpi.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/memory/MemoryEventStorageSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/memory/MemoryEventStorageSpi.java new file mode 100644 index 0000000..7cf2790 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/memory/MemoryEventStorageSpi.java @@ -0,0 +1,272 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.eventstorage.memory; + +import org.apache.ignite.*; +import org.apache.ignite.events.*; +import org.apache.ignite.lang.*; +import org.apache.ignite.resources.*; +import org.apache.ignite.spi.*; +import org.apache.ignite.spi.eventstorage.*; +import org.gridgain.grid.util.typedef.*; +import org.gridgain.grid.util.typedef.internal.*; +import org.jdk8.backport.*; + +import java.util.*; + +import static org.apache.ignite.events.IgniteEventType.*; + +/** + * In-memory {@link org.apache.ignite.spi.eventstorage.EventStorageSpi} implementation. All events are + * kept in the FIFO queue. If no configuration is provided a default expiration + * {@link #DFLT_EXPIRE_AGE_MS} and default count {@link #DFLT_EXPIRE_COUNT} will + * be used. + * <p> + * It's recommended not to set huge size and unlimited TTL because this might + * lead to consuming a lot of memory and result in {@link OutOfMemoryError}. + * Both event expiration time and maximum queue size could be changed at + * runtime. + * <p> + * <h1 class="header">Configuration</h1> + * <h2 class="header">Mandatory</h2> + * This SPI has no mandatory configuration parameters. + * <h2 class="header">Optional</h2> + * The following configuration parameters are optional: + * <ul> + * <li>Event queue size (see {@link #setExpireCount(long)})</li> + * <li>Event time-to-live value (see {@link #setExpireAgeMs(long)})</li> + * <li>{@link #setFilter(org.apache.ignite.lang.IgnitePredicate)} - Event filter that should be used for decision to accept event.</li> + * </ul> + * <h2 class="header">Java Example</h2> + * GridMemoryEventStorageSpi is used by default and should be explicitly configured only + * if some SPI configuration parameters need to be overridden. Examples below insert own + * events queue size value that differs from default 10000. + * <pre name="code" class="java"> + * GridMemoryEventStorageSpi = new GridMemoryEventStorageSpi(); + * + * // Init own events size. + * spi.setExpireCount(2000); + * + * GridConfiguration cfg = new GridConfiguration(); + * + * // Override default event storage SPI. + * cfg.setEventStorageSpi(spi); + * + * // Starts grid. + * G.start(cfg); + * </pre> + * <h2 class="header">Spring Example</h2> + * GridMemoryEventStorageSpi can be configured from Spring XML configuration file: + * <pre name="code" class="xml"> + * <bean id="grid.custom.cfg" class="org.gridgain.grid.GridConfiguration" singleton="true"> + * ... + * <property name="discoverySpi"> + * <bean class="org.gridgain.grid.spi.eventStorage.memory.GridMemoryEventStorageSpi"> + * <property name="expireCount" value="2000"/> + * </bean> + * </property> + * ... + * </bean> + * </pre> + * <p> + * <img src="http://www.gridgain.com/images/spring-small.png";> + * <br> + * For information about Spring framework visit <a href="http://www.springframework.org/";>www.springframework.org</a> + * @see org.apache.ignite.spi.eventstorage.EventStorageSpi + */ +@IgniteSpiMultipleInstancesSupport(true) +public class MemoryEventStorageSpi extends IgniteSpiAdapter implements EventStorageSpi, + MemoryEventStorageSpiMBean { + /** Default event time to live value in milliseconds (value is {@link Long#MAX_VALUE}). */ + public static final long DFLT_EXPIRE_AGE_MS = Long.MAX_VALUE; + + /** Default expire count (value is {@code 10000}). */ + public static final int DFLT_EXPIRE_COUNT = 10000; + + /** */ + @IgniteLoggerResource + private IgniteLogger log; + + /** Event time-to-live value in milliseconds. */ + private long expireAgeMs = DFLT_EXPIRE_AGE_MS; + + /** Maximum queue size. */ + private long expireCnt = DFLT_EXPIRE_COUNT; + + /** Events queue. */ + private ConcurrentLinkedDeque8<IgniteEvent> evts = new ConcurrentLinkedDeque8<>(); + + /** Configured event predicate filter. */ + private IgnitePredicate<IgniteEvent> filter; + + /** + * Gets filter for events to be recorded. + * + * @return Filter to use. + */ + public IgnitePredicate<IgniteEvent> getFilter() { + return filter; + } + + /** + * Sets filter for events to be recorded. + * + * @param filter Filter to use. + */ + @IgniteSpiConfiguration(optional = true) + public void setFilter(IgnitePredicate<IgniteEvent> filter) { + this.filter = filter; + } + + /** {@inheritDoc} */ + @Override public void spiStart(String gridName) throws IgniteSpiException { + // Start SPI start stopwatch. + startStopwatch(); + + assertParameter(expireCnt > 0, "expireCnt > 0"); + assertParameter(expireAgeMs > 0, "expireAgeMs > 0"); + + // Ack parameters. + if (log.isDebugEnabled()) { + log.debug(configInfo("expireAgeMs", expireAgeMs)); + log.debug(configInfo("expireCnt", expireCnt)); + } + + registerMBean(gridName, this, MemoryEventStorageSpiMBean.class); + + // Ack ok start. + if (log.isDebugEnabled()) + log.debug(startInfo()); + } + + /** {@inheritDoc} */ + @Override public void spiStop() throws IgniteSpiException { + unregisterMBean(); + + // Reset events. + evts.clear(); + + // Ack ok stop. + if (log.isDebugEnabled()) + log.debug(stopInfo()); + } + + /** + * Sets events expiration time. All events that exceed this value + * will be removed from the queue when next event comes. + * <p> + * If not provided, default value is {@link #DFLT_EXPIRE_AGE_MS}. + * + * @param expireAgeMs Expiration time in milliseconds. + */ + @IgniteSpiConfiguration(optional = true) + public void setExpireAgeMs(long expireAgeMs) { + this.expireAgeMs = expireAgeMs; + } + + /** + * Sets events queue size. Events will be filtered out when new request comes. + * <p> + * If not provided, default value {@link #DFLT_EXPIRE_COUNT} will be used. + * + * @param expireCnt Maximum queue size. + */ + @IgniteSpiConfiguration(optional = true) + public void setExpireCount(long expireCnt) { + this.expireCnt = expireCnt; + } + + /** {@inheritDoc} */ + @Override public long getExpireAgeMs() { + return expireAgeMs; + } + + /** {@inheritDoc} */ + @Override public long getExpireCount() { + return expireCnt; + } + + /** {@inheritDoc} */ + @Override public long getQueueSize() { + return evts.sizex(); + } + + /** {@inheritDoc} */ + @Override public void clearAll() { + evts.clear(); + } + + /** {@inheritDoc} */ + @Override public <T extends IgniteEvent> Collection<T> localEvents(IgnitePredicate<T> p) { + A.notNull(p, "p"); + + cleanupQueue(); + + return F.retain((Collection<T>)evts, true, p); + } + + /** {@inheritDoc} */ + @Override public void record(IgniteEvent evt) throws IgniteSpiException { + assert evt != null; + + // Filter out events. + if (filter == null || filter.apply(evt)) { + cleanupQueue(); + + evts.add(evt); + + // Make sure to filter out metrics updates to prevent log from flooding. + if (evt.type() != EVT_NODE_METRICS_UPDATED && log.isDebugEnabled()) + log.debug("Event recorded: " + evt); + } + } + + /** + * Method cleans up all events that either outnumber queue size + * or exceeds time-to-live value. It does none if someone else + * cleans up queue (lock is locked) or if there are queue readers + * (readersNum > 0). + */ + private void cleanupQueue() { + long now = U.currentTimeMillis(); + + long queueOversize = evts.sizex() - expireCnt; + + for (int i = 0; i < queueOversize && evts.sizex() > expireCnt; i++) { + IgniteEvent expired = evts.poll(); + + if (log.isDebugEnabled()) + log.debug("Event expired by count: " + expired); + } + + while (true) { + ConcurrentLinkedDeque8.Node<IgniteEvent> node = evts.peekx(); + + if (node == null) // Queue is empty. + break; + + IgniteEvent evt = node.item(); + + if (evt == null) // Competing with another thread. + continue; + + if (now - evt.timestamp() < expireAgeMs) + break; + + if (evts.unlinkx(node) && log.isDebugEnabled()) + log.debug("Event expired by age: " + node.item()); + } + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(MemoryEventStorageSpi.class, this); + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/memory/MemoryEventStorageSpiMBean.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/memory/MemoryEventStorageSpiMBean.java b/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/memory/MemoryEventStorageSpiMBean.java new file mode 100644 index 0000000..e3d249f --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/memory/MemoryEventStorageSpiMBean.java @@ -0,0 +1,57 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.eventstorage.memory; + +import org.apache.ignite.mbean.*; +import org.apache.ignite.spi.*; + +/** + * Management bean for {@link MemoryEventStorageSpi}. + * Beside properties defined for every SPI bean this one gives access to: + * <ul> + * <li>Event expiration time (see {@link #getExpireAgeMs()})</li> + * <li>Maximum queue size (see {@link #getExpireCount()})</li> + * <li>Method that removes all items from queue (see {@link #clearAll()})</li> + * </ul> + */ +@IgniteMBeanDescription("MBean that provides access to memory event storage SPI configuration.") +public interface MemoryEventStorageSpiMBean extends IgniteSpiManagementMBean { + /** + * Gets event time-to-live value. Implementation must guarantee + * that event would not be accessible if its lifetime exceeds this value. + * + * @return Event time-to-live. + */ + @IgniteMBeanDescription("Event time-to-live value.") + public long getExpireAgeMs(); + + /** + * Gets maximum event queue size. New incoming events will oust + * oldest ones if queue size exceeds this limit. + * + * @return Maximum event queue size. + */ + @IgniteMBeanDescription("Maximum event queue size.") + public long getExpireCount(); + + /** + * Gets current queue size of the event queue. + * + * @return Current queue size of the event queue. + */ + @IgniteMBeanDescription("Current event queue size.") + public long getQueueSize(); + + /** + * Removes all events from the event queue. + */ + @IgniteMBeanDescription("Removes all events from the event queue.") + public void clearAll(); +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/memory/package.html ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/memory/package.html b/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/memory/package.html new file mode 100644 index 0000000..7b8a251 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/memory/package.html @@ -0,0 +1,15 @@ +<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd";> +<!-- + @html.file.header + _________ _____ __________________ _____ + __ ____/___________(_)______ /__ ____/______ ____(_)_______ + _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ +--> +<html> +<body> + <!-- Package description. --> + Contains <b>default</b> in-memory implementation for event storage SPI. +</body> +</html> http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/package.html ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/package.html b/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/package.html new file mode 100644 index 0000000..83f8dd9 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/eventstorage/package.html @@ -0,0 +1,15 @@ +<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd";> +<!-- + @html.file.header + _________ _____ __________________ _____ + __ ____/___________(_)______ /__ ____/______ ____(_)_______ + _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ +--> +<html> +<body> + <!-- Package description. --> + Contains APIs for event storage SPI. +</body> +</html> http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/failover/FailoverContext.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/failover/FailoverContext.java b/modules/core/src/main/java/org/apache/ignite/spi/failover/FailoverContext.java new file mode 100644 index 0000000..d66ec56 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/failover/FailoverContext.java @@ -0,0 +1,47 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.failover; + +import org.apache.ignite.cluster.*; +import org.apache.ignite.compute.*; +import org.gridgain.grid.*; +import org.gridgain.grid.spi.loadbalancing.*; +import java.util.*; + +/** + * This interface defines a set of operations available to failover SPI + * one a given failed job. + */ +public interface FailoverContext { + /** + * Gets current task session. + * + * @return Grid task session. + */ + public ComputeTaskSession getTaskSession(); + + /** + * Gets failed result of job execution. + * + * @return Result of a failed job. + */ + public ComputeJobResult getJobResult(); + + /** + * Gets the next balanced node for failed job. Internally this method will + * delegate to load balancing SPI (see {@link GridLoadBalancingSpi} to + * determine the optimal node for execution. + * + * @param top Topology to pick balanced node from. + * @return The next balanced node. + * @throws GridException If anything failed. + */ + public ClusterNode getBalancedNode(List<ClusterNode> top) throws GridException; +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/failover/FailoverSpi.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/failover/FailoverSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/failover/FailoverSpi.java new file mode 100644 index 0000000..e593c87 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/failover/FailoverSpi.java @@ -0,0 +1,60 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.failover; + +import org.apache.ignite.cluster.*; +import org.apache.ignite.spi.*; + +import java.util.*; + +/** + * Failover SPI provides developer with ability to supply custom logic for handling + * failed execution of a grid job. Job execution can fail for a number of reasons: + * <ul> + * <li>Job execution threw an exception (runtime, assertion or error)</li> + * <li>Node on which job was execution left topology (crashed or stopped)</li> + * <li>Collision SPI on remote node cancelled a job before it got a chance to execute (job rejection).</li> + * </ul> + * In all cases failover SPI takes failed job (as failover context) and list of all + * grid nodes and provides another node on which the job execution will be retried. + * It is up to failover SPI to make sure that job is not mapped to the node it + * failed on. The failed node can be retrieved from + * {@link org.apache.ignite.compute.ComputeJobResult#getNode() GridFailoverContext.getJobResult().node()} + * method. + * <p> + * GridGain comes with the following built-in failover SPI implementations: + * <ul> + * <li>{@link org.apache.ignite.spi.failover.never.NeverFailoverSpi}</li> + * <li>{@link org.apache.ignite.spi.failover.always.AlwaysFailoverSpi}</li> + * <li>{@link org.apache.ignite.spi.failover.jobstealing.JobStealingFailoverSpi}</li> + * </ul> + * <b>NOTE:</b> this SPI (i.e. methods in this interface) should never be used directly. SPIs provide + * internal view on the subsystem and is used internally by GridGain kernal. In rare use cases when + * access to a specific implementation of this SPI is required - an instance of this SPI can be obtained + * via {@link org.apache.ignite.Ignite#configuration()} method to check its configuration properties or call other non-SPI + * methods. Note again that calling methods from this interface on the obtained instance can lead + * to undefined behavior and explicitly not supported. + */ +public interface FailoverSpi extends IgniteSpi { + /** + * This method is called when method {@link org.apache.ignite.compute.ComputeTask#result(org.apache.ignite.compute.ComputeJobResult, List)} returns + * value {@link org.apache.ignite.compute.ComputeJobResultPolicy#FAILOVER} policy indicating that the result of + * job execution must be failed over. Implementation of this method should examine failover + * context and choose one of the grid nodes from supplied {@code topology} to retry job execution + * on it. For best performance it is advised that {@link FailoverContext#getBalancedNode(List)} + * method is used to select node for execution of failed job. + * + * @param ctx Failover context. + * @param top Collection of all grid nodes within task topology (may include failed node). + * @return New node to route this job to or {@code null} if new node cannot be picked. + * If job failover fails (returns {@code null}) the whole task will be failed. + */ + public ClusterNode failover(FailoverContext ctx, List<ClusterNode> top); +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/failover/always/AlwaysFailoverSpi.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/failover/always/AlwaysFailoverSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/failover/always/AlwaysFailoverSpi.java new file mode 100644 index 0000000..7a8e591 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/failover/always/AlwaysFailoverSpi.java @@ -0,0 +1,238 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.failover.always; + +import org.apache.ignite.*; +import org.apache.ignite.cluster.*; +import org.apache.ignite.resources.*; +import org.apache.ignite.spi.*; +import org.gridgain.grid.*; +import org.apache.ignite.spi.failover.*; +import org.gridgain.grid.util.typedef.*; +import org.gridgain.grid.util.typedef.internal.*; + +import java.util.*; + +/** + * Failover SPI that always reroutes a failed job to another node. + * Note, that at first an attempt will be made to reroute the failed job + * to a node that was not part of initial split for a better chance of + * success. If no such nodes are available, then an attempt will be made to + * reroute the failed job to the nodes in the initial split minus the node + * the job is failed on. If none of the above attempts succeeded, then the + * job will not be failed over and {@code null} will be returned. + * <p> + * <h1 class="header">Configuration</h1> + * This SPI is default failover SPI and does not have to be explicitly + * configured unless configuration parameters need to be changed. + * <h2 class="header">Mandatory</h2> + * This SPI has no mandatory configuration parameters. + * <h2 class="header">Optional</h2> + * This SPI has following optional configuration parameters: + * <ul> + * <li> + * Maximum failover attempts for a single job (see {@link #setMaximumFailoverAttempts(int)}). + * If maximum failover attempts is reached, then job will not be failed-over and, + * hence, will fail. + * </li> + * </ul> + * Here is a Java example how to configure grid with {@code GridAlwaysFailoverSpi} failover SPI. + * <pre name="code" class="java"> + * GridAlwaysFailoverSpi spi = new GridAlwaysFailoverSpi(); + * + * // Override maximum failover attempts. + * spi.setMaximumFailoverAttempts(5); + * + * GridConfiguration cfg = new GridConfiguration(); + * + * // Override default failover SPI. + * cfg.setFailoverSpiSpi(spi); + * + * // Starts grid. + * G.start(cfg); + * </pre> + * Here is an example of how to configure {@code GridAlwaysFailoverSpi} from Spring XML configuration file. + * <pre name="code" class="xml"> + * <property name="failoverSpi"> + * <bean class="org.gridgain.grid.spi.failover.always.GridAlwaysFailoverSpi"> + * <property name="maximumFailoverAttempts" value="5"/> + * </bean> + * </property> + * </pre> + * <p> + * <img src="http://www.gridgain.com/images/spring-small.png";> + * <br> + * For information about Spring framework visit <a href="http://www.springframework.org/";>www.springframework.org</a> + * @see org.apache.ignite.spi.failover.FailoverSpi + */ +@IgniteSpiMultipleInstancesSupport(true) +@IgniteSpiConsistencyChecked(optional = true) +public class AlwaysFailoverSpi extends IgniteSpiAdapter implements FailoverSpi, AlwaysFailoverSpiMBean { + /** Maximum number of attempts to execute a failed job on another node (default is {@code 5}). */ + public static final int DFLT_MAX_FAILOVER_ATTEMPTS = 5; + + /** + * Name of job context attribute containing all nodes a job failed on. + * + * @see org.apache.ignite.compute.ComputeJobContext + */ + public static final String FAILED_NODE_LIST_ATTR = "gg:failover:failednodelist"; + + /** Maximum attempts attribute key should be the same on all nodes. */ + public static final String MAX_FAILOVER_ATTEMPT_ATTR = "gg:failover:maxattempts"; + + /** Injected grid logger. */ + @IgniteLoggerResource + private IgniteLogger log; + + /** Maximum number of attempts to execute a failed job on another node. */ + private int maxFailoverAttempts = DFLT_MAX_FAILOVER_ATTEMPTS; + + /** Number of jobs that were failed over. */ + private int totalFailoverJobs; + + /** {@inheritDoc} */ + @Override public int getMaximumFailoverAttempts() { + return maxFailoverAttempts; + } + + /** + * Sets maximum number of attempts to execute a failed job on another node. + * If not specified, {@link #DFLT_MAX_FAILOVER_ATTEMPTS} value will be used. + * + * @param maxFailoverAttempts Maximum number of attempts to execute a failed job on another node. + */ + @IgniteSpiConfiguration(optional = true) + public void setMaximumFailoverAttempts(int maxFailoverAttempts) { + this.maxFailoverAttempts = maxFailoverAttempts; + } + + /** {@inheritDoc} */ + @Override public int getTotalFailoverJobsCount() { + return totalFailoverJobs; + } + + /** {@inheritDoc} */ + @Override public Map<String, Object> getNodeAttributes() throws IgniteSpiException { + return F.<String, Object>asMap(createSpiAttributeName(MAX_FAILOVER_ATTEMPT_ATTR), maxFailoverAttempts); + } + + /** {@inheritDoc} */ + @Override public void spiStart(String gridName) throws IgniteSpiException { + // Start SPI start stopwatch. + startStopwatch(); + + assertParameter(maxFailoverAttempts >= 0, "maxFailoverAttempts >= 0"); + + if (log.isDebugEnabled()) + log.debug(configInfo("maximumFailoverAttempts", maxFailoverAttempts)); + + registerMBean(gridName, this, AlwaysFailoverSpiMBean.class); + + // Ack ok start. + if (log.isDebugEnabled()) + log.debug(startInfo()); + } + + /** {@inheritDoc} */ + @Override public void spiStop() throws IgniteSpiException { + unregisterMBean(); + + // Ack ok stop. + if (log.isDebugEnabled()) + log.debug(stopInfo()); + } + + /** {@inheritDoc} */ + @SuppressWarnings("unchecked") + @Override public ClusterNode failover(FailoverContext ctx, List<ClusterNode> top) { + assert ctx != null; + assert top != null; + + if (log.isDebugEnabled()) + log.debug("Received failed job result: " + ctx.getJobResult()); + + if (top.isEmpty()) { + U.warn(log, "Received empty topology for failover and is forced to fail."); + + // Nowhere to failover to. + return null; + } + + Collection<UUID> failedNodes = ctx.getJobResult().getJobContext().getAttribute(FAILED_NODE_LIST_ATTR); + + if (failedNodes == null) + failedNodes = U.newHashSet(1); + + Integer failoverCnt = failedNodes.size(); + + if (failoverCnt >= maxFailoverAttempts) { + U.warn(log, "Job failover failed because number of maximum failover attempts is exceeded [failedJob=" + + ctx.getJobResult().getJob() + ", maxFailoverAttempts=" + maxFailoverAttempts + ']'); + + return null; + } + + failedNodes.add(ctx.getJobResult().getNode().id()); + + // Copy. + List<ClusterNode> newTop = new ArrayList<>(top.size()); + + for (ClusterNode node : top) + if (!failedNodes.contains(node.id())) + newTop.add(node); + + if (newTop.isEmpty()) { + U.warn(log, "Received topology with only nodes that job had failed on (forced to fail) [failedNodes=" + + failedNodes + ']'); + + // Nowhere to failover to. + return null; + } + + try { + ClusterNode node = ctx.getBalancedNode(newTop); + + if (node == null) + U.warn(log, "Load balancer returned null node for topology: " + newTop); + else { + // Increment failover count. + ctx.getJobResult().getJobContext().setAttribute(FAILED_NODE_LIST_ATTR, failedNodes); + + totalFailoverJobs++; + } + + if (node != null) + U.warn(log, "Failed over job to a new node [newNode=" + node.id() + + ", oldNode=" + ctx.getJobResult().getNode().id() + + ", sesId=" + ctx.getTaskSession().getId() + + ", job=" + ctx.getJobResult().getJob() + + ", jobCtx=" + ctx.getJobResult().getJobContext() + + ", task=" + ctx.getTaskSession().getTaskName() + ']'); + + return node; + } + catch (GridException e) { + U.error(log, "Failed to get next balanced node for failover: " + ctx, e); + + return null; + } + } + + /** {@inheritDoc} */ + @Override protected List<String> getConsistentAttributeNames() { + return Collections.singletonList(createSpiAttributeName(MAX_FAILOVER_ATTEMPT_ATTR)); + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(AlwaysFailoverSpi.class, this); + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/failover/always/AlwaysFailoverSpiMBean.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/failover/always/AlwaysFailoverSpiMBean.java b/modules/core/src/main/java/org/apache/ignite/spi/failover/always/AlwaysFailoverSpiMBean.java new file mode 100644 index 0000000..1fb4c59 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/failover/always/AlwaysFailoverSpiMBean.java @@ -0,0 +1,36 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.failover.always; + +import org.apache.ignite.mbean.*; +import org.apache.ignite.spi.*; + +/** + * Management bean for {@link AlwaysFailoverSpi}. + */ +@IgniteMBeanDescription("MBean that provides access to always failover SPI configuration.") +public interface AlwaysFailoverSpiMBean extends IgniteSpiManagementMBean { + /** + * Gets maximum number of attempts to execute a failed job on another node. + * If not specified, {@link AlwaysFailoverSpi#DFLT_MAX_FAILOVER_ATTEMPTS} value will be used. + * + * @return Maximum number of attempts to execute a failed job on another node. + */ + @IgniteMBeanDescription("Maximum number of attempts to execute a failed job on another node.") + public int getMaximumFailoverAttempts(); + + /** + * Get total number of jobs that were failed over. + * + * @return Total number of failed over jobs. + */ + @IgniteMBeanDescription("Total number of jobs that were failed over.") + public int getTotalFailoverJobsCount(); +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/failover/always/package.html ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/failover/always/package.html b/modules/core/src/main/java/org/apache/ignite/spi/failover/always/package.html new file mode 100644 index 0000000..6299303 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/failover/always/package.html @@ -0,0 +1,15 @@ +<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd";> +<!-- + @html.file.header + _________ _____ __________________ _____ + __ ____/___________(_)______ /__ ____/______ ____(_)_______ + _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ +--> +<html> +<body> + <!-- Package description. --> + Contains <b>default</b> "always" failover SPI. +</body> +</html> http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/failover/jobstealing/JobStealingFailoverSpi.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/failover/jobstealing/JobStealingFailoverSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/failover/jobstealing/JobStealingFailoverSpi.java new file mode 100644 index 0000000..45c5f3d --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/failover/jobstealing/JobStealingFailoverSpi.java @@ -0,0 +1,343 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.failover.jobstealing; + +import org.apache.ignite.*; +import org.apache.ignite.cluster.*; +import org.apache.ignite.resources.*; +import org.apache.ignite.spi.*; +import org.gridgain.grid.*; +import org.apache.ignite.spi.failover.*; +import org.gridgain.grid.util.typedef.*; +import org.gridgain.grid.util.typedef.internal.*; + +import java.util.*; + +import static org.apache.ignite.spi.collision.jobstealing.JobStealingCollisionSpi.*; + +/** + * Job stealing failover SPI needs to always be used in conjunction with + * {@link org.apache.ignite.spi.collision.jobstealing.JobStealingCollisionSpi} SPI. When {@link org.apache.ignite.spi.collision.jobstealing.JobStealingCollisionSpi} + * receives a <b>steal</b> request and rejects jobs so they can be routed to the + * appropriate node, it is the responsibility of this {@code GridJobStealingFailoverSpi} + * SPI to make sure that the job is indeed re-routed to the node that has sent the initial + * request to <b>steal</b> it. + * <p> + * {@code GridJobStealingFailoverSpi} knows where to route a job based on the + * {@link org.apache.ignite.spi.collision.jobstealing.JobStealingCollisionSpi#THIEF_NODE_ATTR} job context attribute (see {@link org.apache.ignite.compute.ComputeJobContext}). + * Prior to rejecting a job, {@link org.apache.ignite.spi.collision.jobstealing.JobStealingCollisionSpi} will populate this + * attribute with the ID of the node that wants to <b>steal</b> this job. + * Then {@code GridJobStealingFailoverSpi} will read the value of this attribute and + * route the job to the node specified. + * <p> + * If failure is caused by a node crash, and not by <b>steal</b> request, then this + * SPI behaves identically to {@link org.apache.ignite.spi.failover.always.AlwaysFailoverSpi}, and tries to find the + * next balanced node to fail-over a job to. + * <p> + * <h1 class="header">Configuration</h1> + * <h2 class="header">Mandatory</h2> + * This SPI has no mandatory configuration parameters. + * <h2 class="header">Optional</h2> + * This SPI has following optional configuration parameters: + * <ul> + * <li>Maximum failover attempts for a single job (see {@link #setMaximumFailoverAttempts(int)}).</li> + * </ul> + * Here is a Java example on how to configure grid with {@code GridJobStealingFailoverSpi}. + * <pre name="code" class="java"> + * GridJobStealingFailoverSpi spi = new GridJobStealingFailoverSpi(); + * + * // Override maximum failover attempts. + * spi.setMaximumFailoverAttempts(5); + * + * GridConfiguration cfg = new GridConfiguration(); + * + * // Override default failover SPI. + * cfg.setFailoverSpiSpi(spi); + * + * // Starts grid. + * G.start(cfg); + </pre> + * Here is an example of how to configure {@code GridJobStealingFailoverSpi} from Spring XML configuration file. + * <pre name="code" class="xml"> + * <property name="failoverSpi"> + * <bean class="org.gridgain.grid.spi.failover.jobstealing.GridJobStealingFailoverSpi"> + * <property name="maximumFailoverAttempts" value="5"/> + * </bean> + * </property> + * </pre> + * <p> + * <img src="http://www.gridgain.com/images/spring-small.png";> + * <br> + * For information about Spring framework visit <a href="http://www.springframework.org/";>www.springframework.org</a> + * @see org.apache.ignite.spi.failover.FailoverSpi + */ +@IgniteSpiMultipleInstancesSupport(true) +@IgniteSpiConsistencyChecked(optional = true) +public class JobStealingFailoverSpi extends IgniteSpiAdapter implements FailoverSpi, + JobStealingFailoverSpiMBean { + /** Maximum number of attempts to execute a failed job on another node (default is {@code 5}). */ + public static final int DFLT_MAX_FAILOVER_ATTEMPTS = 5; + + /** + * Name of job context attribute containing all nodes a job failed on. Note + * that this list does not include nodes that a job was stolen from. + * + * @see org.apache.ignite.compute.ComputeJobContext + */ + static final String FAILED_NODE_LIST_ATTR = "gg:failover:failednodelist"; + + /** + * Name of job context attribute containing current failover attempt count. + * This count is incremented every time the same job gets failed over to + * another node for execution if it was not successfully stolen. + * + * @see org.apache.ignite.compute.ComputeJobContext + */ + static final String FAILOVER_ATTEMPT_COUNT_ATTR = "gg:failover:attemptcount"; + + /** Maximum failover attempts job context attribute name. */ + private static final String MAX_FAILOVER_ATTEMPT_ATTR = "gg:failover:maxattempts"; + + /** Injected grid logger. */ + @IgniteLoggerResource + private IgniteLogger log; + + /** Maximum number of attempts to execute a failed job on another node. */ + private int maxFailoverAttempts = DFLT_MAX_FAILOVER_ATTEMPTS; + + /** Number of jobs that were failed over. */ + private int totalFailedOverJobs; + + /** Number of jobs that were stolen. */ + private int totalStolenJobs; + + /** {@inheritDoc} */ + @Override public int getMaximumFailoverAttempts() { + return maxFailoverAttempts; + } + + /** + * Sets maximum number of attempts to execute a failed job on another node. + * If job gets stolen and thief node exists then it is not considered as + * failed job. + * If not specified, {@link #DFLT_MAX_FAILOVER_ATTEMPTS} value will be used. + * <p> + * Note this value must be identical for all grid nodes in the grid. + * + * @param maxFailoverAttempts Maximum number of attempts to execute a failed + * job on another node. + */ + @IgniteSpiConfiguration(optional = true) + public void setMaximumFailoverAttempts(int maxFailoverAttempts) { + this.maxFailoverAttempts = maxFailoverAttempts; + } + + /** {@inheritDoc} */ + @Override public int getTotalFailedOverJobsCount() { + return totalFailedOverJobs; + } + + /** {@inheritDoc} */ + @Override public int getTotalStolenJobsCount() { + return totalStolenJobs; + } + + /** {@inheritDoc} */ + @Override public Map<String, Object> getNodeAttributes() throws IgniteSpiException { + return F.<String, Object>asMap(createSpiAttributeName(MAX_FAILOVER_ATTEMPT_ATTR), maxFailoverAttempts); + } + + /** {@inheritDoc} */ + @Override public void spiStart(String gridName) throws IgniteSpiException { + // Start SPI start stopwatch. + startStopwatch(); + + assertParameter(maxFailoverAttempts >= 0, "maximumFailoverAttempts >= 0"); + + if (log.isDebugEnabled()) + log.debug(configInfo("maxFailoverAttempts", maxFailoverAttempts)); + + registerMBean(gridName, this, JobStealingFailoverSpiMBean.class); + + // Ack ok start. + if (log.isDebugEnabled()) + log.debug(startInfo()); + } + + /** {@inheritDoc} */ + @Override public void spiStop() throws IgniteSpiException { + unregisterMBean(); + + // Ack ok stop. + if (log.isDebugEnabled()) + log.debug(stopInfo()); + } + + /** {@inheritDoc} */ + @SuppressWarnings("unchecked") + @Override public ClusterNode failover(FailoverContext ctx, List<ClusterNode> top) { + assert ctx != null; + assert top != null; + + if (top.isEmpty()) { + U.warn(log, "Received empty subgrid and is forced to fail."); + + // Nowhere to failover to. + return null; + } + + Integer failoverCnt = ctx.getJobResult().getJobContext().getAttribute(FAILOVER_ATTEMPT_COUNT_ATTR); + + if (failoverCnt == null) + failoverCnt = 0; + + if (failoverCnt > maxFailoverAttempts) { + U.error(log, "Failover count exceeded maximum failover attempts parameter [failedJob=" + + ctx.getJobResult().getJob() + ", maxFailoverAttempts=" + maxFailoverAttempts + ']'); + + return null; + } + + if (failoverCnt == maxFailoverAttempts) { + U.warn(log, "Job failover failed because number of maximum failover attempts is exceeded [failedJob=" + + ctx.getJobResult().getJob() + ", maxFailoverAttempts=" + maxFailoverAttempts + ']'); + + return null; + } + + try { + ClusterNode thief = null; + boolean isNodeFailed = false; + + UUID thiefId = ctx.getJobResult().getJobContext().getAttribute(THIEF_NODE_ATTR); + + if (thiefId != null) { + // Clear attribute. + ctx.getJobResult().getJobContext().setAttribute(THIEF_NODE_ATTR, null); + + thief = getSpiContext().node(thiefId); + + if (thief != null) { + // If sender != receiver. + if (thief.equals(ctx.getJobResult().getNode())) { + U.error(log, "Job stealer node is equal to job node (will fail-over using " + + "load-balancing): " + thief.id()); + + isNodeFailed = true; + + thief = null; + } + else if (!top.contains(thief)) { + U.warn(log, "Thief node is not part of task topology (will fail-over using load-balancing) " + + "[thief=" + thiefId + ", topSize=" + top.size() + ']'); + + thief = null; + } + + if (log.isDebugEnabled()) + log.debug("Failing-over stolen job [from=" + ctx.getJobResult().getNode() + ", to=" + + thief + ']'); + } + else { + isNodeFailed = true; + + U.warn(log, "Thief node left grid (will fail-over using load balancing): " + thiefId); + } + } + else + isNodeFailed = true; + + // If job was not stolen or stolen node is not part of topology, + // then failover the regular way. + if (thief == null) { + Collection<UUID> failedNodes = ctx.getJobResult().getJobContext().getAttribute(FAILED_NODE_LIST_ATTR); + + if (failedNodes == null) + failedNodes = U.newHashSet(1); + + if (isNodeFailed) + failedNodes.add(ctx.getJobResult().getNode().id()); + + // Set updated failed node set into job context. + ctx.getJobResult().getJobContext().setAttribute(FAILED_NODE_LIST_ATTR, failedNodes); + + // Copy. + List<ClusterNode> newTop = new ArrayList<>(top.size()); + + for (ClusterNode n : top) { + // Add non-failed nodes to topology. + if (!failedNodes.contains(n.id())) + newTop.add(n); + } + + if (newTop.isEmpty()) { + U.warn(log, "Received topology with only nodes that job had failed on (forced to fail) " + + "[failedNodes=" + failedNodes + ']'); + + // Nowhere to failover to. + return null; + } + + thief = ctx.getBalancedNode(newTop); + + if (thief == null) + U.warn(log, "Load balancer returned null node for topology: " + newTop); + } + + if (isNodeFailed) + // This is a failover, not stealing. + failoverCnt++; + + // Even if it was stealing and thief node left grid we assume + // that it is failover because of the fail. + ctx.getJobResult().getJobContext().setAttribute(FAILOVER_ATTEMPT_COUNT_ATTR, failoverCnt); + + if (thief != null) { + totalFailedOverJobs++; + + if (isNodeFailed) { + U.warn(log, "Failed over job to a new node [newNode=" + thief.id() + + ", oldNode=" + ctx.getJobResult().getNode().id() + + ", sesId=" + ctx.getTaskSession().getId() + + ", job=" + ctx.getJobResult().getJob() + + ", jobCtx=" + ctx.getJobResult().getJobContext() + + ", task=" + ctx.getTaskSession().getTaskName() + ']'); + } + else { + totalStolenJobs++; + if (log.isInfoEnabled()) + log.info("Stealing job to a new node [newNode=" + thief.id() + + ", oldNode=" + ctx.getJobResult().getNode().id() + + ", sesId=" + ctx.getTaskSession().getId() + + ", job=" + ctx.getJobResult().getJob() + + ", jobCtx=" + ctx.getJobResult().getJobContext() + + ", task=" + ctx.getTaskSession().getTaskName() + ']'); + } + } + + return thief; + } + catch (GridException e) { + U.error(log, "Failed to get next balanced node for failover: " + ctx, e); + + return null; + } + } + + /** {@inheritDoc} */ + @Override protected List<String> getConsistentAttributeNames() { + return Collections.singletonList(createSpiAttributeName(MAX_FAILOVER_ATTEMPT_ATTR)); + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(JobStealingFailoverSpi.class, this); + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/failover/jobstealing/JobStealingFailoverSpiMBean.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/failover/jobstealing/JobStealingFailoverSpiMBean.java b/modules/core/src/main/java/org/apache/ignite/spi/failover/jobstealing/JobStealingFailoverSpiMBean.java new file mode 100644 index 0000000..39ddbb6 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/failover/jobstealing/JobStealingFailoverSpiMBean.java @@ -0,0 +1,46 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.failover.jobstealing; + +import org.apache.ignite.mbean.*; +import org.apache.ignite.spi.*; + +/** + * Management bean for {@link JobStealingFailoverSpi}. + */ +@IgniteMBeanDescription("MBean that provides access to job stealing failover SPI configuration.") +public interface JobStealingFailoverSpiMBean extends IgniteSpiManagementMBean { + /** + * Gets maximum number of attempts to execute a failed job on another node. + * If job gets stolen and thief node exists then it is not considered as + * failed job. + * If not specified, {@link JobStealingFailoverSpi#DFLT_MAX_FAILOVER_ATTEMPTS} value will be used. + * + * @return Maximum number of attempts to execute a failed job on another node. + */ + @IgniteMBeanDescription("Maximum number of attempts to execute a failed job on another node.") + public int getMaximumFailoverAttempts(); + + /** + * Get total number of jobs that were failed over including stolen ones. + * + * @return Total number of failed over jobs. + */ + @IgniteMBeanDescription("Total number of jobs that were failed over including stolen ones.") + public int getTotalFailedOverJobsCount(); + + /** + * Get total number of jobs that were stolen. + * + * @return Total number of stolen jobs. + */ + @IgniteMBeanDescription("Total number of jobs that were stolen.") + public int getTotalStolenJobsCount(); +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/failover/jobstealing/package.html ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/failover/jobstealing/package.html b/modules/core/src/main/java/org/apache/ignite/spi/failover/jobstealing/package.html new file mode 100644 index 0000000..ebb77c2 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/failover/jobstealing/package.html @@ -0,0 +1,15 @@ +<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd";> +<!-- + @html.file.header + _________ _____ __________________ _____ + __ ____/___________(_)______ /__ ____/______ ____(_)_______ + _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ +--> +<html> +<body> + <!-- Package description. --> + Contains job stealing failover SPI. +</body> +</html> http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/failover/never/NeverFailoverSpi.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/failover/never/NeverFailoverSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/failover/never/NeverFailoverSpi.java new file mode 100644 index 0000000..eb38726 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/failover/never/NeverFailoverSpi.java @@ -0,0 +1,94 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.failover.never; + +import org.apache.ignite.*; +import org.apache.ignite.cluster.*; +import org.apache.ignite.resources.*; +import org.apache.ignite.spi.*; +import org.apache.ignite.spi.failover.*; +import org.gridgain.grid.util.typedef.internal.*; +import java.util.*; + +/** + * This class provides failover SPI implementation that never fails over. This implementation + * never fails over a failed job by always returning {@code null} out of + * {@link org.apache.ignite.spi.failover.FailoverSpi#failover(org.apache.ignite.spi.failover.FailoverContext, List)} method. + * <h1 class="header">Configuration</h1> + * <h2 class="header">Mandatory</h2> + * This SPI has no mandatory configuration parameters. + * <h2 class="header">Optional</h2> + * This SPI has no optional configuration parameters. + * <p> + * Here is a Java example on how to configure grid with {@code GridNeverFailoverSpi}: + * <pre name="code" class="java"> + * GridNeverFailoverSpi spi = new GridNeverFailoverSpi(); + * + * GridConfiguration cfg = new GridConfiguration(); + * + * // Override default failover SPI. + * cfg.setFailoverSpiSpi(spi); + * + * // Starts grid. + * G.start(cfg); + * </pre> + * Here is an example on how to configure grid with {@code GridNeverFailoverSpi} from Spring XML configuration file: + * <pre name="code" class="xml"> + * <property name="failoverSpi"> + * <bean class="org.gridgain.grid.spi.failover.never.GridNeverFailoverSpi"/> + * </property> + * </pre> + * <p> + * <img src="http://www.gridgain.com/images/spring-small.png";> + * <br> + * For information about Spring framework visit <a href="http://www.springframework.org/";>www.springframework.org</a> + * @see org.apache.ignite.spi.failover.FailoverSpi + */ +@IgniteSpiMultipleInstancesSupport(true) +public class NeverFailoverSpi extends IgniteSpiAdapter implements FailoverSpi, NeverFailoverSpiMBean { + /** Injected grid logger. */ + @IgniteLoggerResource + private IgniteLogger log; + + /** {@inheritDoc} */ + @Override public void spiStart(String gridName) throws IgniteSpiException { + // Start SPI start stopwatch. + startStopwatch(); + + registerMBean(gridName, this, NeverFailoverSpiMBean.class); + + // Ack ok start. + if (log.isDebugEnabled()) + log.debug(startInfo()); + } + + /** {@inheritDoc} */ + @Override public void spiStop() throws IgniteSpiException { + unregisterMBean(); + + // Ack ok stop. + if (log.isDebugEnabled()) + log.debug(stopInfo()); + } + + /** {@inheritDoc} */ + @Override public ClusterNode failover(FailoverContext ctx, List<ClusterNode> top) { + U.warn(log, "Returning 'null' node for failed job (failover will not happen) [job=" + + ctx.getJobResult().getJob() + ", task=" + ctx.getTaskSession().getTaskName() + + ", sessionId=" + ctx.getTaskSession().getId() + ']'); + + return null; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(NeverFailoverSpi.class, this); + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/failover/never/NeverFailoverSpiMBean.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/failover/never/NeverFailoverSpiMBean.java b/modules/core/src/main/java/org/apache/ignite/spi/failover/never/NeverFailoverSpiMBean.java new file mode 100644 index 0000000..ceb5fe7 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/failover/never/NeverFailoverSpiMBean.java @@ -0,0 +1,21 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.failover.never; + +import org.apache.ignite.mbean.*; +import org.apache.ignite.spi.*; + +/** + * Management bean for {@link NeverFailoverSpi}. + */ +@IgniteMBeanDescription("MBean that provides access to never failover SPI configuration.") +public interface NeverFailoverSpiMBean extends IgniteSpiManagementMBean { + // No-op. +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/failover/never/package.html ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/failover/never/package.html b/modules/core/src/main/java/org/apache/ignite/spi/failover/never/package.html new file mode 100644 index 0000000..81e38aa --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/failover/never/package.html @@ -0,0 +1,15 @@ +<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd";> +<!-- + @html.file.header + _________ _____ __________________ _____ + __ ____/___________(_)______ /__ ____/______ ____(_)_______ + _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ +--> +<html> +<body> + <!-- Package description. --> + Contains "never" failover SPI. +</body> +</html> http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/apache/ignite/spi/failover/package.html ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/failover/package.html b/modules/core/src/main/java/org/apache/ignite/spi/failover/package.html new file mode 100644 index 0000000..3b28cd2 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/failover/package.html @@ -0,0 +1,15 @@ +<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd";> +<!-- + @html.file.header + _________ _____ __________________ _____ + __ ____/___________(_)______ /__ ____/______ ____(_)_______ + _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ +--> +<html> +<body> + <!-- Package description. --> + Contains APIs for failover SPI. +</body> +</html> http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/gridgain/grid/kernal/GridGainEx.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/GridGainEx.java b/modules/core/src/main/java/org/gridgain/grid/kernal/GridGainEx.java index 716c435..4a0774d 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/GridGainEx.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/GridGainEx.java @@ -43,10 +43,10 @@ import org.apache.ignite.spi.deployment.local.*; import org.apache.ignite.spi.discovery.*; import org.apache.ignite.spi.discovery.tcp.*; import org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.*; -import org.gridgain.grid.spi.eventstorage.*; -import org.gridgain.grid.spi.eventstorage.memory.*; -import org.gridgain.grid.spi.failover.*; -import org.gridgain.grid.spi.failover.always.*; +import org.apache.ignite.spi.eventstorage.*; +import org.apache.ignite.spi.eventstorage.memory.*; +import org.apache.ignite.spi.failover.*; +import org.apache.ignite.spi.failover.always.*; import org.gridgain.grid.spi.indexing.*; import org.gridgain.grid.spi.loadbalancing.*; import org.gridgain.grid.spi.loadbalancing.roundrobin.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/gridgain/grid/kernal/managers/eventstorage/GridEventStorageManager.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/managers/eventstorage/GridEventStorageManager.java b/modules/core/src/main/java/org/gridgain/grid/kernal/managers/eventstorage/GridEventStorageManager.java index b595b6c..db12d53 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/managers/eventstorage/GridEventStorageManager.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/managers/eventstorage/GridEventStorageManager.java @@ -20,7 +20,7 @@ import org.gridgain.grid.kernal.managers.*; import org.gridgain.grid.kernal.managers.communication.*; import org.gridgain.grid.kernal.managers.deployment.*; import org.gridgain.grid.security.*; -import org.gridgain.grid.spi.eventstorage.*; +import org.apache.ignite.spi.eventstorage.*; import org.gridgain.grid.util.*; import org.gridgain.grid.util.future.*; import org.gridgain.grid.util.typedef.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/gridgain/grid/kernal/managers/failover/GridFailoverContextImpl.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/managers/failover/GridFailoverContextImpl.java b/modules/core/src/main/java/org/gridgain/grid/kernal/managers/failover/GridFailoverContextImpl.java index a63dad9..5e78803 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/managers/failover/GridFailoverContextImpl.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/managers/failover/GridFailoverContextImpl.java @@ -14,7 +14,7 @@ import org.apache.ignite.compute.*; import org.gridgain.grid.*; import org.gridgain.grid.kernal.*; import org.gridgain.grid.kernal.managers.loadbalancer.*; -import org.gridgain.grid.spi.failover.*; +import org.apache.ignite.spi.failover.*; import org.gridgain.grid.util.typedef.internal.*; import org.gridgain.grid.util.tostring.*; import java.util.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/gridgain/grid/kernal/managers/failover/GridFailoverManager.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/managers/failover/GridFailoverManager.java b/modules/core/src/main/java/org/gridgain/grid/kernal/managers/failover/GridFailoverManager.java index b4fbde0..97e1679 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/managers/failover/GridFailoverManager.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/managers/failover/GridFailoverManager.java @@ -14,7 +14,7 @@ import org.apache.ignite.compute.*; import org.gridgain.grid.*; import org.gridgain.grid.kernal.*; import org.gridgain.grid.kernal.managers.*; -import org.gridgain.grid.spi.failover.*; +import org.apache.ignite.spi.failover.*; import java.util.*; /** http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/725526fa/modules/core/src/main/java/org/gridgain/grid/spi/eventstorage/EventStorageSpi.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/spi/eventstorage/EventStorageSpi.java b/modules/core/src/main/java/org/gridgain/grid/spi/eventstorage/EventStorageSpi.java deleted file mode 100644 index da21ba9..0000000 --- a/modules/core/src/main/java/org/gridgain/grid/spi/eventstorage/EventStorageSpi.java +++ /dev/null @@ -1,57 +0,0 @@ -/* @java.file.header */ - -/* _________ _____ __________________ _____ - * __ ____/___________(_)______ /__ ____/______ ____(_)_______ - * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ - * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / - * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ - */ - -package org.gridgain.grid.spi.eventstorage; - -import org.apache.ignite.events.*; -import org.apache.ignite.lang.*; -import org.apache.ignite.spi.*; - -import java.util.*; - -/** - * This SPI provides local node events storage. SPI allows for recording local - * node events and querying recorded local events. Every node during its life-cycle - * goes through a serious of events such as task deployment, task execution, job - * execution, etc. For - * performance reasons GridGain is designed to store all locally produced events - * locally. These events can be later retrieved using either distributed query: - * <ul> - * <li>{@link org.apache.ignite.IgniteEvents#remoteQuery(org.apache.ignite.lang.IgnitePredicate, long, int...)}</li> - * </ul> - * or local only query: - * <ul> - * <li>{@link org.apache.ignite.IgniteEvents#localQuery(org.apache.ignite.lang.IgnitePredicate, int...)}</li> - * </ul> - * <b>NOTE:</b> this SPI (i.e. methods in this interface) should never be used directly. SPIs provide - * internal view on the subsystem and is used internally by GridGain kernal. In rare use cases when - * access to a specific implementation of this SPI is required - an instance of this SPI can be obtained - * via {@link org.apache.ignite.Ignite#configuration()} method to check its configuration properties or call other non-SPI - * methods. Note again that calling methods from this interface on the obtained instance can lead - * to undefined behavior and explicitly not supported. - * @see org.apache.ignite.events.IgniteEvent - */ -public interface EventStorageSpi extends IgniteSpi { - /** - * Queries locally-stored events only. Events could be filtered out - * by given predicate filter. - * - * @param p Event predicate filter. - * @return Collection of events. - */ - public <T extends IgniteEvent> Collection<T> localEvents(IgnitePredicate<T> p); - - /** - * Records single event. - * - * @param evt Event that should be recorded. - * @throws org.apache.ignite.spi.IgniteSpiException If event recording failed for any reason. - */ - public void record(IgniteEvent evt) throws IgniteSpiException; -}
![http://www.gridgain.com/images/spring-small.png"](http://www.gridgain.com/images/spring-small.png"){kind=link}