| 88 | import org.slf4j.LoggerFactory; |
| 89 | |
| 90 | public abstract class Executor implements Callable, JCQueue.Consumer { |
| 91 | |
| 92 | private static final Logger LOG = LoggerFactory.getLogger(Executor.class); |
| 93 | |
| 94 | protected final WorkerState workerData; |
| 95 | protected final WorkerTopologyContext workerTopologyContext; |
| 96 | protected final List<Long> executorId; |
| 97 | protected final List<Integer> taskIds; |
| 98 | protected final String componentId; |
| 99 | protected final AtomicBoolean openOrPrepareWasCalled; |
| 100 | protected final Map<String, Object> topoConf; |
| 101 | protected final Map<String, Object> conf; |
| 102 | protected final String stormId; |
| 103 | protected final HashMap sharedExecutorData; |
| 104 | protected final CountDownLatch workerReady; |
| 105 | protected final AtomicBoolean stormActive; |
| 106 | protected final AtomicReference<Map<String, DebugOptions>> stormComponentDebug; |
| 107 | protected final Runnable suicideFn; |
| 108 | protected final IStormClusterState stormClusterState; |
| 109 | protected final Map<Integer, String> taskToComponent; |
| 110 | protected final Map<Integer, Map<Integer, Map<String, IMetric>>> intervalToTaskToMetricToRegistry; |
| 111 | protected final Map<String, Map<String, LoadAwareCustomStreamGrouping>> streamToComponentToGrouper; |
| 112 | protected final List<LoadAwareCustomStreamGrouping> groupers; |
| 113 | protected final ReportErrorAndDie reportErrorDie; |
| 114 | protected final BooleanSupplier sampler; |
| 115 | protected final String type; |
| 116 | protected final IReportError reportError; |
| 117 | protected final Random rand; |
| 118 | protected final JCQueue receiveQueue; |
| 119 | protected final Map<String, String> credentials; |
| 120 | protected final Boolean isDebug; |
| 121 | protected final Boolean hasEventLoggers; |
| 122 | protected final boolean ackingEnabled; |
| 123 | protected final MpscChunkedArrayQueue<AddressedTuple> pendingEmits = new MpscChunkedArrayQueue<>(1024, (int) Math.pow(2, 30)); |
| 124 | private final AddressedTuple flushTuple; |
| 125 | protected ExecutorTransfer executorTransfer; |
| 126 | protected ArrayList<Task> idToTask; |
| 127 | protected int idToTaskBase; |
| 128 | protected String hostname; |
| 129 | private static final double msDurationFactor = 1.0 / TimeUnit.MILLISECONDS.toNanos(1); |
| 130 | private AtomicBoolean needToRefreshCreds = new AtomicBoolean(false); |
| 131 | private final RateCounter reportedErrorCount; |
| 132 | private final boolean enableV2MetricsDataPoints; |
| 133 | private final Integer v2MetricsTickInterval; |
| 134 | |
| 135 | protected Executor(WorkerState workerData, List<Long> executorId, Map<String, String> credentials, String type) { |
| 136 | this.workerData = workerData; |
| 137 | this.executorId = executorId; |
| 138 | this.type = type; |
| 139 | this.workerTopologyContext = workerData.getWorkerTopologyContext(); |
| 140 | this.taskIds = StormCommon.executorIdToTasks(executorId); |
| 141 | this.componentId = workerTopologyContext.getComponentId(taskIds.get(0)); |
| 142 | this.openOrPrepareWasCalled = new AtomicBoolean(false); |
| 143 | this.topoConf = normalizedComponentConf(workerData.getTopologyConf(), workerTopologyContext, componentId); |
| 144 | this.receiveQueue = (workerData.getExecutorReceiveQueueMap().get(executorId)); |
| 145 | this.stormId = workerData.getTopologyId(); |
| 146 | this.conf = workerData.getConf(); |
| 147 | this.sharedExecutorData = new HashMap(); |