Hazelcast Jet ProcessorTasklet Sınıfı - Processor'ı Çalıştırır

Giriş

Şu satırı dahil ederiz

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

Tasklet arayüzünden kalıtır. Üye alanlar şöyle

public class ProcessorTasklet implements Tasklet {

 private Processor processor;
  ...
}

private void stateMachineStep metodu

Tasket'in FSM gerçekleştirimi bu metod içinde. Bu metod iç içe çağrılabilir. Bir exception şöyle

at com.hazelcast.jet.impl.connector.WriteJdbcP.process(WriteJdbcP.java:194)
at com.hazelcast.jet.impl.processor.ProcessorWrapper.process(ProcessorWrapper.java:97)
at com.hazelcast.jet.impl.pipeline.FunctionAdapter$AdaptingProcessor.process(FunctionAdapter.java:226)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.lambda$processInbox$2f647568$2(ProcessorTasklet.java:490)
at com.hazelcast.jet.function.RunnableEx.run(RunnableEx.java:31)
at com.hazelcast.jet.impl.util.Util.doWithClassLoader(Util.java:532)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.processInbox(ProcessorTasklet.java:490)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:341)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:336)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:328)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.call(ProcessorTasklet.java:291)
at com.hazelcast.jet.impl.execution.TaskletExecutionService$BlockingWorker.run(TaskletExecutionService.java:315)
... 5 more

COMPLETE

Kod şöyle. Yani processor nesnesini çalıştırır

case COMPLETE:
  complete();
  return;

CLOSE

Kod şöyle. Yani processor nesnesini kapatır

case CLOSE:
  if (isCooperative() && !processor.closeIsCooperative()) {
    if (closeFuture == null) {
      ClassLoader contextCl = Thread.currentThread().getContextClassLoader();
      closeFuture = executionService.submit(() -> doWithClassLoader(contextCl, 
        this::closeProcessor));
      progTracker.madeProgress();
    }
    if (!closeFuture.isDone()) {
      return;
    }
    progTracker.madeProgress();
  } else {
    closeProcessor();
  }
  state = END;
  progTracker.done();
  return;