Hazelcast Jet ExecutionContext Sınıfı

Giriş

Şu satırı dahil ederiz 

import com.hazelcast.jet.impl.execution.ExecutionContext;

Kod şöyle.

public class ExecutionContext implements DynamicMetricsProvider {
  ...
  private final long jobId;
  private final long executionId;
  private final boolean isLightJob;
  private volatile Address coordinator;
  private volatile Set<Address> participants;
  private final long createdOn = System.nanoTime();
  private final Object executionLock = new Object();
  private final ILogger logger;
  private final Counter startTime = MwCounter.newMwCounter(-1);
  private final Counter completionTime = MwCounter.newMwCounter(-1);

  // key: resource identifier
  // we use ConcurrentHashMap because ConcurrentMap doesn't guarantee
  //that computeIfAbsent
  // executes the supplier strictly only if it's needed.
  private final ConcurrentHashMap<String, File> tempDirectories 
    = new ConcurrentHashMap<>();

  private String jobName;

  private volatile Map<SenderReceiverKey, ReceiverTasklet> receiverMap;
  private volatile Map<SenderReceiverKey, SenderTasklet> senderMap;
  private final Map<SenderReceiverKey, Queue<byte[]>> receiverQueuesMap;

  private List<VertexDef> vertices = emptyList();
  private List<Tasklet> tasklets = emptyList();

  // future which is completed only after all tasklets are completed 
  // and contains execution result
  private volatile CompletableFuture<Void> executionFuture;

  // future which can only be used to cancel the local execution.
  private final CompletableFuture<Void> cancellationFuture = 
      new CompletableFuture<>();

  private final NodeEngineImpl nodeEngine;
  private final JetServiceBackend jetServiceBackend;
  private volatile SnapshotContext snapshotContext;
  private JobConfig jobConfig;

  private boolean metricsEnabled;
  private volatile RawJobMetrics jobMetrics = RawJobMetrics.empty();

  private InternalSerializationService serializationService;
  private final AtomicBoolean executionCompleted = new AtomicBoolean();
  ...
}

tasklet listesinin atanması için çağrı sırası şöyle

InitExecutionOperation.doRun
  JobExecutionService.runLightJob
    ExecutionContext.initialize
      ExecutionContext.initWithPlan

beginExecution metodu

Kod şöyle. List<Tasklet> tasklets nesnesini TaskletExecutionService nesnesine verir ve çalıştırmasını ister

/**
  * Starts local execution of job by submitting tasklets to execution service. If
  * execution was cancelled earlier then execution will not be started.
  *
  * Returns a future which is completed only when all tasklets are completed. If
  * execution was already cancelled before this method is called then the returned
  * future is completed immediately. The future returned can't be cancelled,
  * instead terminateExecution should be used.
  */
public CompletableFuture<Void> beginExecution(TaskletExecutionService taskletExecService)

initWithPlan metodu

tasklet listesinin atanması için çağrı sırası şöyle

InitExecutionOperation.doRun
  JobExecutionService.runLightJob
    ExecutionContext.initialize
      ExecutionContext.initWithPlan

Kod şöyle

ExecutionPlan plan = ...;
tasklets = plan.getTasklets();