Package com.hazelcast.jet.core

Interface ProcessorMetaSupplier

All Superinterfaces:

Serializable

All Known Implementing Classes:

ProcessorMetaSupplier.RandomMemberPms, ProcessorMetaSupplier.SpecificMemberPms

Functional Interface:

This is a functional interface and can therefore be used as the assignment target for a lambda expression or method reference.

@FunctionalInterface
public interface ProcessorMetaSupplier
extends Serializable

Factory of ProcessorSupplier instances. The starting point of the chain leading to the eventual creation of Processor instances on each cluster member:

1. client or member creates ProcessorMetaSupplier as a part of the DAG;
2. client or member sends it to the job coordinator (if the member is job coordinator and the DAG consists of reusable ProcessorMetaSuppliers, the serialization may be skipped);
3. the job coordinator uses it to create one ProcessorSupplier for each cluster member;
4. serializes each ProcessorSupplier and sends it to its target member;
5. the target member deserializes and uses it to instantiate as many instances of Processor as requested by the parallelism property on the corresponding Vertex.

Before being asked to create ProcessorSuppliers this meta-supplier will be given access to the Hazelcast instance and, in particular, its cluster topology and partitioning services. It can use the information from these services to precisely parameterize each Processor instance that will be created on each member.

Since:

Jet 3.0

Nested Class Summary

Nested Classes

Modifier and Type	Interface	Description
static interface	ProcessorMetaSupplier.Context	Context passed to the meta-supplier at init time on the member that received a job request from the client.
static class	ProcessorMetaSupplier.ExpectNothingProcessorSupplier
static class	ProcessorMetaSupplier.RandomMemberPms
static class	ProcessorMetaSupplier.SpecificMemberPms	A meta-supplier that will only use the given ProcessorSupplier on a node with given Address or random member if address is not given.

Method Summary

Modifier and Type	Method	Description
default void	close(Throwable error)	Called on coordinator member after execution has finished on all members, successfully or not.
default boolean	closeIsCooperative()	Returns true if the close(Throwable) method of this instance is cooperative.
static ProcessorMetaSupplier	forceTotalParallelismOne(ProcessorSupplier supplier)	Variant of forceTotalParallelismOne(ProcessorSupplier, String, Permission) where the node for the supplier will be chosen randomly and without any required permission.
static ProcessorMetaSupplier	forceTotalParallelismOne(ProcessorSupplier supplier, Address memberAddress)	Wraps the provided ProcessorSupplier into a meta-supplier that will only use the given ProcessorSupplier on a node with the given Address.
static ProcessorMetaSupplier	forceTotalParallelismOne(ProcessorSupplier supplier, String partitionKey)	Variant of forceTotalParallelismOne(ProcessorSupplier, String, Permission) without any required permission.
static ProcessorMetaSupplier	forceTotalParallelismOne(ProcessorSupplier supplier, String partitionKey, Permission permission)	Wraps the provided ProcessorSupplier into a meta-supplier that will only use the given ProcessorSupplier on a single node.
static ProcessorMetaSupplier	forceTotalParallelismOne(ProcessorSupplier supplier, Permission permission)	Variant of forceTotalParallelismOne(ProcessorSupplier, String, Permission) where the node for the supplier will be chosen randomly.
Function<? super Address,? extends ProcessorSupplier>	get(List<Address> addresses)	Called to create a mapping from member Address to the ProcessorSupplier that will be sent to that member.
static Address	getOwnerAddress(ProcessorMetaSupplier.Context context, Object partitionKey)	Since the partition owner may not participate in job execution, always ProcessorMetaSupplier.Context.partitionAssignment() should be used to find the "actual" owner.
default Permission	getRequiredPermission()	Returns the required permission to execute the vertex which has this ProcessorMetaSupplier.
default Map<String,String>	getTags()	Returns the metadata on this supplier, a string-to-string map.
default void	init(ProcessorMetaSupplier.Context context)	Called on the cluster member that receives the job request.
default boolean	initIsCooperative()	Returns true if both the init(Context) and get(List) methods of this instance are cooperative.
default boolean	isReusable()	Returns true if the same instance can be reused in different job executions or in different vertices.
static ProcessorMetaSupplier	of(int preferredLocalParallelism, FunctionEx<? super Address,? extends ProcessorSupplier> addressToSupplier)	Factory method that creates a ProcessorMetaSupplier from the supplied function that maps a cluster member address to a ProcessorSupplier.
static ProcessorMetaSupplier	of(int preferredLocalParallelism, SupplierEx<? extends Processor> procSupplier)	Factory method that wraps the given Supplier<Processor> and uses it as the supplier of all Processor instances.
static ProcessorMetaSupplier	of(int preferredLocalParallelism, ProcessorSupplier procSupplier)	Variant of of(int, Permission, ProcessorSupplier) where the processor does not require any permission to run.
static ProcessorMetaSupplier	of(int preferredLocalParallelism, Permission permission, ProcessorSupplier procSupplier)	Factory method that wraps the given ProcessorSupplier and returns the same instance for each given Address.
static ProcessorMetaSupplier	of(int preferredLocalParallelism, Permission permission, ProcessorSupplier procSupplier, String connectorName)	Factory method that wraps the given ProcessorSupplier and returns the same instance for each given Address.
static ProcessorMetaSupplier	of(FunctionEx<? super Address,? extends ProcessorSupplier> addressToSupplier)	Factory method that creates a ProcessorMetaSupplier from the supplied function that maps a cluster member address to a ProcessorSupplier.
static ProcessorMetaSupplier	of(SupplierEx<? extends Processor> procSupplier)	Factory method that wraps the given Supplier<Processor> and uses it as the supplier of all Processor instances.
static ProcessorMetaSupplier	of(ProcessorSupplier procSupplier)	Variant of of(Permission, ProcessorSupplier) where the processor does not require any permission to run.
static ProcessorMetaSupplier	of(Permission permission, ProcessorSupplier procSupplier)	Wraps the provided ProcessorSupplier into a meta-supplier that will always return it.
static ProcessorMetaSupplier	preferLocalParallelismOne(SupplierEx<? extends Processor> procSupplier)	Variant of preferLocalParallelismOne(ProcessorSupplier) where the supplied SupplierEx<Processor> will be wrapped into a ProcessorSupplier.
static ProcessorMetaSupplier	preferLocalParallelismOne(ProcessorSupplier supplier)	Wraps the provided ProcessorSupplier into a meta-supplier that will always return it.
static ProcessorMetaSupplier	preferLocalParallelismOne(Permission permission, SupplierEx<? extends Processor> procSupplier)	Variant of preferLocalParallelismOne(SupplierEx) where the processor requires given permission to run.
static ProcessorMetaSupplier	preferLocalParallelismOne(Permission permission, SupplierEx<? extends Processor> procSupplier, String connectorName)	Variant of preferLocalParallelismOne(SupplierEx) where the processor requires given permission to run.
static ProcessorMetaSupplier	preferLocalParallelismOne(Permission permission, ProcessorSupplier supplier)	Variant of preferLocalParallelismOne(ProcessorSupplier) where the processor requires given permission to run.
default int	preferredLocalParallelism()	Returns the local parallelism the vertex should be configured with.
static ProcessorMetaSupplier	randomMember(ProcessorSupplier supplier)	Wraps the provided ProcessorSupplier into a meta-supplier that will only use the given ProcessorSupplier on a random node

Method Details

getRequiredPermission

@Nullable
default Permission getRequiredPermission()

Returns the required permission to execute the vertex which has this ProcessorMetaSupplier. This is an Enterprise feature.

getTags

@Nonnull
default Map<String,String> getTags()

Returns the metadata on this supplier, a string-to-string map. There is no predefined metadata; this facility exists to allow the DAG vertices to contribute some information to the execution planning phase.

Since:

Jet 4.0

preferredLocalParallelism

default int preferredLocalParallelism()

Returns the local parallelism the vertex should be configured with. The default implementation returns Vertex.LOCAL_PARALLELISM_USE_DEFAULT.

init

default void init(@Nonnull
 ProcessorMetaSupplier.Context context)
           throws Exception

Called on the cluster member that receives the job request. Gives access to the Hazelcast instance's services and provides the parallelism parameters determined from the cluster size.

Throws:

Exception

See Also:

• isReusable()

initIsCooperative

default boolean initIsCooperative()

Returns true if both the init(Context) and get(List) methods of this instance are cooperative. If they are not, the call to the init() and get() method is off-loaded to another thread.

Since:

5.2

get

@Nonnull
Function<? super Address,? extends ProcessorSupplier> get(@Nonnull
 List<Address> addresses)

Called to create a mapping from member Address to the ProcessorSupplier that will be sent to that member. Jet calls this method with a list of all cluster members' addresses and the returned function must be a mapping that returns a non-null value for each given address.

The method will be called once per job execution on the job's coordinator member. init() will have already been called.

See Also:

• isReusable()

closeIsCooperative

default boolean closeIsCooperative()

Returns true if the close(Throwable) method of this instance is cooperative. If it's not, the call to the close() method is off-loaded to another thread.

Since:

5.2

close

default void close(@Nullable
 Throwable error)
            throws Exception

Called on coordinator member after execution has finished on all members, successfully or not. This method will be called after ProcessorSupplier.close(Throwable) has been called on all available members. The job can be restarted later.

If there is an exception during the creation of the execution plan, this method will be called regardless of whether the init() or get() method have been called or not. If this method throws an exception, it will be logged and ignored; it won't be reported as a job failure.

If you rely on the fact that this method is run once per cluster, it can happen that it is not called at all, if the coordinator member crashed. It can be also called multiple times, if the job restarts.

Parameters:

error - the exception (if any) that caused the job to fail; null in the case of successful job completion. Note that it might not be the actual error that caused the job to fail - it can be several other exceptions. We only guarantee that it's non-null if the job didn't complete successfully.

Throws:

Exception

See Also:

• isReusable()

isReusable

default boolean isReusable()

Returns true if the same instance can be reused in different job executions or in different vertices. In that case, init(com.hazelcast.jet.core.ProcessorMetaSupplier.Context), get(java.util.List<com.hazelcast.cluster.Address>) and close(java.lang.Throwable) methods must be thread-safe and obey additional conditions defined below.

When a job is submitted from a client, the job definition is serialized, so the job coordinator will receive a different copy of processor meta-suppliers even if they are used multiple times within the same DAG, or across different DAGs, or submitted through different jobs. While this serialization mechanism ensures that processor meta-supplier instances do not share any internal state, it is unnecessary —and avoided— for jobs consisting of reusable meta-suppliers and submitted from the job coordinator —which is always the case for member-originated light jobs.

Non-reusable meta-suppliers (default) have a simple order of method executions: init(com.hazelcast.jet.core.ProcessorMetaSupplier.Context) (once), get(java.util.List<com.hazelcast.cluster.Address>) (at most once after init(com.hazelcast.jet.core.ProcessorMetaSupplier.Context), close(java.lang.Throwable) (last).

Reusable meta-suppliers differ because the meta supplier instance may be shared and reused. That is why:

1. init(com.hazelcast.jet.core.ProcessorMetaSupplier.Context) can be invoked multiple times (also after close(java.lang.Throwable)).
2. get(java.util.List<com.hazelcast.cluster.Address>) can be invoked multiple times, but each get(java.util.List<com.hazelcast.cluster.Address>) invocation will be preceded by init(com.hazelcast.jet.core.ProcessorMetaSupplier.Context) invocation for given job execution.
3. close(java.lang.Throwable) can be invoked multiple times with or without preceding invocations of the other methods.
4. Meta-supplier method invocation sequences for different concurrent job executions may be interleaved.

It is recommended that reusable meta-supplier does not have any mutable state that is changed by any of the methods. It is, however, allowed to initialize some thread-safe constant fields in init(com.hazelcast.jet.core.ProcessorMetaSupplier.Context) method, for example, save some constant data from ProcessorMetaSupplier.Context for further usage. Note, however, that cluster topology may change and should not be stored.

Since:

5.3

of

@Nonnull
static ProcessorMetaSupplier of(int preferredLocalParallelism,
 @Nullable
 Permission permission,
 @Nonnull
 ProcessorSupplier procSupplier)

Factory method that wraps the given ProcessorSupplier and returns the same instance for each given Address.

Parameters:

preferredLocalParallelism - the value to return from preferredLocalParallelism()

permission - the required permission to run the processor

procSupplier - the processor supplier

of

@Nonnull
static ProcessorMetaSupplier of(int preferredLocalParallelism,
 @Nullable
 Permission permission,
 @Nonnull
 ProcessorSupplier procSupplier,
 @Nullable
 String connectorName)

Factory method that wraps the given ProcessorSupplier and returns the same instance for each given Address.

Parameters:

preferredLocalParallelism - the value to return from preferredLocalParallelism()

permission - the required permission to run the processor

procSupplier - the processor supplier

connectorName - connector name

of

@Nonnull
static ProcessorMetaSupplier of(int preferredLocalParallelism,
 @Nonnull
 ProcessorSupplier procSupplier)

Variant of of(int, Permission, ProcessorSupplier) where the processor does not require any permission to run.

of

@Nonnull
static ProcessorMetaSupplier of(@Nullable
 Permission permission,
 @Nonnull
 ProcessorSupplier procSupplier)

Wraps the provided ProcessorSupplier into a meta-supplier that will always return it. The preferredLocalParallelism() of the meta-supplier will be Vertex.LOCAL_PARALLELISM_USE_DEFAULT.

of

@Nonnull
static ProcessorMetaSupplier of(@Nonnull
 ProcessorSupplier procSupplier)

Variant of of(Permission, ProcessorSupplier) where the processor does not require any permission to run.

of

@Nonnull
static ProcessorMetaSupplier of(int preferredLocalParallelism,
 @Nonnull
 SupplierEx<? extends Processor> procSupplier)

Factory method that wraps the given Supplier<Processor> and uses it as the supplier of all Processor instances. Specifically, returns a meta-supplier that will always return the result of calling ProcessorSupplier.of(SupplierEx).

Parameters:

preferredLocalParallelism - the value to return from preferredLocalParallelism()

procSupplier - the supplier of processors

of

@Nonnull
static ProcessorMetaSupplier of(@Nonnull
 SupplierEx<? extends Processor> procSupplier)

Factory method that wraps the given Supplier<Processor> and uses it as the supplier of all Processor instances. Specifically, returns a meta-supplier that will always return the result of calling ProcessorSupplier.of(SupplierEx). The preferredLocalParallelism() of the meta-supplier will be Vertex.LOCAL_PARALLELISM_USE_DEFAULT.

of

@Nonnull
static ProcessorMetaSupplier of(int preferredLocalParallelism,
 @Nonnull
 FunctionEx<? super Address,? extends ProcessorSupplier> addressToSupplier)

Factory method that creates a ProcessorMetaSupplier from the supplied function that maps a cluster member address to a ProcessorSupplier.

Parameters:

preferredLocalParallelism - the value to return from preferredLocalParallelism()

addressToSupplier - the mapping from address to ProcessorSupplier

of

@Nonnull
static ProcessorMetaSupplier of(@Nonnull
 FunctionEx<? super Address,? extends ProcessorSupplier> addressToSupplier)

Factory method that creates a ProcessorMetaSupplier from the supplied function that maps a cluster member address to a ProcessorSupplier. The preferredLocalParallelism() of the meta-supplier will be Vertex.LOCAL_PARALLELISM_USE_DEFAULT.

preferLocalParallelismOne

@Nonnull
static ProcessorMetaSupplier preferLocalParallelismOne(@Nonnull
 ProcessorSupplier supplier)

Wraps the provided ProcessorSupplier into a meta-supplier that will always return it. The preferredLocalParallelism() of the meta-supplier will be one, i.e., no local parallelization.

The parallelism will be overridden if the Vertex.localParallelism(int) is set to a specific value.

preferLocalParallelismOne

@Nonnull
static ProcessorMetaSupplier preferLocalParallelismOne(@Nullable
 Permission permission,
 @Nonnull
 ProcessorSupplier supplier)

Variant of preferLocalParallelismOne(ProcessorSupplier) where the processor requires given permission to run.

preferLocalParallelismOne

@Nonnull
static ProcessorMetaSupplier preferLocalParallelismOne(@Nonnull
 SupplierEx<? extends Processor> procSupplier)

Variant of preferLocalParallelismOne(ProcessorSupplier) where the supplied SupplierEx<Processor> will be wrapped into a ProcessorSupplier.

preferLocalParallelismOne

@Nonnull
static ProcessorMetaSupplier preferLocalParallelismOne(@Nullable
 Permission permission,
 @Nonnull
 SupplierEx<? extends Processor> procSupplier)

Variant of preferLocalParallelismOne(SupplierEx) where the processor requires given permission to run.

preferLocalParallelismOne

@Nonnull
static ProcessorMetaSupplier preferLocalParallelismOne(@Nullable
 Permission permission,
 @Nonnull
 SupplierEx<? extends Processor> procSupplier,
 @Nullable
 String connectorName)

Variant of preferLocalParallelismOne(SupplierEx) where the processor requires given permission to run.

forceTotalParallelismOne

@Nonnull
static ProcessorMetaSupplier forceTotalParallelismOne(@Nonnull
 ProcessorSupplier supplier,
 @Nullable
 Permission permission)

Variant of forceTotalParallelismOne(ProcessorSupplier, String, Permission) where the node for the supplier will be chosen randomly.

forceTotalParallelismOne

@Nonnull
static ProcessorMetaSupplier forceTotalParallelismOne(@Nonnull
 ProcessorSupplier supplier)

Variant of forceTotalParallelismOne(ProcessorSupplier, String, Permission) where the node for the supplier will be chosen randomly and without any required permission.

forceTotalParallelismOne

@Nonnull
static ProcessorMetaSupplier forceTotalParallelismOne(@Nonnull
 ProcessorSupplier supplier,
 @Nonnull
 String partitionKey)

Variant of forceTotalParallelismOne(ProcessorSupplier, String, Permission) without any required permission.

forceTotalParallelismOne

@Nonnull
static ProcessorMetaSupplier forceTotalParallelismOne(@Nonnull
 ProcessorSupplier supplier,
 @Nonnull
 String partitionKey,
 @Nullable
 Permission permission)

Wraps the provided ProcessorSupplier into a meta-supplier that will only use the given ProcessorSupplier on a single node. The node will be chosen according to the partitionKey supplied. This is mainly provided as a convenience for implementing non-distributed sources where data can't be read in parallel by multiple consumers. When used as a sink or intermediate vertex, the DAG should ensure that only the processor instance on the designated node receives any data, otherwise an IllegalStateException will be thrown.

The vertex containing the ProcessorMetaSupplier must have a local parallelism setting of 1, otherwise {code IllegalArgumentException} is thrown.

Parameters:

supplier - the supplier that will be wrapped

partitionKey - the supplier will only be created on the node that owns the supplied partition key

permission - the required permission to run the processor

Returns:

the wrapped ProcessorMetaSupplier

Throws:

IllegalArgumentException - if vertex has local parallelism setting of greater than 1

getOwnerAddress

static Address getOwnerAddress(ProcessorMetaSupplier.Context context,
 Object partitionKey)

Since the partition owner may not participate in job execution, always ProcessorMetaSupplier.Context.partitionAssignment() should be used to find the "actual" owner.

forceTotalParallelismOne

@Nonnull
static ProcessorMetaSupplier forceTotalParallelismOne(@Nonnull
 ProcessorSupplier supplier,
 @Nonnull
 Address memberAddress)

Wraps the provided ProcessorSupplier into a meta-supplier that will only use the given ProcessorSupplier on a node with the given Address. This is mainly provided as a convenience for implementing non-distributed sources where data can't be read in parallel by multiple consumers. When used as a sink or intermediate vertex, the DAG should ensure that only the processor instance on the designated node receives any data, otherwise an IllegalStateException will be thrown.

The vertex containing the ProcessorMetaSupplier must have a local parallelism setting of 1, otherwise {code IllegalArgumentException} is thrown.

Parameters:

supplier - the supplier that will be wrapped

memberAddress - the supplier will only be created on the node with given Address

Returns:

the wrapped ProcessorMetaSupplier

Throws:

IllegalArgumentException - if vertex has local parallelism setting of greater than 1

randomMember

@Nonnull
@Beta
static ProcessorMetaSupplier randomMember(@Nonnull
 ProcessorSupplier supplier)

Wraps the provided ProcessorSupplier into a meta-supplier that will only use the given ProcessorSupplier on a random node

Parameters:

supplier - the supplier that will be wrapped

Returns:

the wrapped ProcessorMetaSupplier

Since:

5.3