import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
@Override
public void init(@Nonnull Outbox outbox, @Nonnull Context context) {
    logger = context.logger();
    outbox = new LoggingOutbox(outbox, peekOutput, peekSnapshot);

    // Fix issue #595: pass a logger with real class name to processor
    // We do this only if context is ProcCtx (that is, not for tests where TestProcessorContext can be used
    // and also other objects could be mocked or null, such as jetInstance())
    if (context instanceof ProcCtx) {
        ProcCtx c = (ProcCtx) context;
        NodeEngine nodeEngine = ((HazelcastInstanceImpl) c.jetInstance().getHazelcastInstance()).node.nodeEngine;
        ILogger newLogger = nodeEngine.getLogger(
                createLoggerName(wrappedProcessor.getClass().getName(), c.vertexName(), c.globalProcessorIndex()));
        context = new ProcCtx(c.jetInstance(), c.getSerializationService(), newLogger, c.vertexName(),
                c.globalProcessorIndex(), c.processingGuarantee());
    }

    wrappedProcessor.init(outbox, context);
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
private static void initPartitionOwnersAndMembers(NodeEngine nodeEngine,
                                                  MembersView membersView,
                                                  Collection<MemberInfo> members,
                                                  Address[] partitionOwners) {
    IPartitionService partitionService = nodeEngine.getPartitionService();
    for (int partitionId = 0; partitionId < partitionOwners.length; partitionId++) {
        Address address = partitionService.getPartitionOwnerOrWait(partitionId);

        MemberInfo member;
        if ((member = membersView.getMember(address)) == null) {
            // Address in partition table doesn't exist in member list,
            // it has just left the cluster.
            throw new TopologyChangedException("Topology changed, " + address + " is not in original member list");
        }

        // add member to known members
        members.add(member);
        partitionOwners[partitionId] = address;
    }
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
@Override
public void init(NodeEngine nodeEngine, Properties properties) {
	logger.info("init; got properties: {}", properties); 
	this.nodeEngine = nodeEngine;
	this.schemaName = properties.getProperty(pn_schema_name);
	this.populationSize = Integer.parseInt(properties.getProperty(pn_schema_population_size));
	String dataPath = properties.getProperty(pn_schema_store_data_path);
	String nodeNum = properties.getProperty(pn_node_instance);
	int buffSize = 2048*100;
	String bSize = properties.getProperty(pn_schema_population_buffer_size);
	if (bSize != null) {
		buffSize = Integer.parseInt(bSize);
	}
	logger.info("init; will open doc store from path: {}; instance: {}; buffer size: {} docs", dataPath, nodeNum, buffSize);
	docStore = new DocumentMemoryStore(dataPath, nodeNum, buffSize);
		
	nodeEngine.getPartitionService().addMigrationListener(this);
	nodeEngine.getHazelcastInstance().getCluster().addMembershipListener(this);
	nodeEngine.getHazelcastInstance().getLifecycleService().addLifecycleListener(this);
	nodeEngine.getHazelcastInstance().getUserContext().put(ctx_popService, this);
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
public void init(final NodeEngine nodeEngine, Properties properties) {
    try {
        super.init(nodeEngine, properties);
    } catch (Exception e) {
        //e.printStackTrace();
    }

    final LockService lockService = nodeEngine.getSharedService(LockService.SERVICE_NAME);
    if (lockService != null) {
        lockService.registerLockStoreConstructor(SERVICE_NAME, new ConstructorFunction<ObjectNamespace, LockStoreInfo>() {
            public LockStoreInfo createNew(final ObjectNamespace key) {
                final MapContainer mapContainer = getMapContainer(key.getObjectName());
                return new LockStoreInfo() {
                    public int getBackupCount() {
                        return mapContainer.getBackupCount();
                    }

                    public int getAsyncBackupCount() {
                        return mapContainer.getAsyncBackupCount();
                    }
                };
            }
        });
    }
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
private <T> ScheduledFuture<T> submitToPartitionOwner(Callable<T> task, int partitionId, long delay, long period, boolean fixedRate) {
        if (task == null) {
            throw new NullPointerException("task can't be null");
        }
        if (isShutdown()) {
            throw new RejectedExecutionException(getRejectionMessage());
        }
        NodeEngine nodeEngine = getNodeEngine();
        Data taskData = nodeEngine.toData(task);
        String uuid = buildRandomUuidString();
        String name = getName();
        ScheduledCallableTaskOperation op = new ScheduledCallableTaskOperation(name, uuid, taskData, delay, period, fixedRate);
        ICompletableFuture future = invoke(partitionId, op);
        return new ScheduledDelegatingFuture<T>(future, nodeEngine.getSerializationService(), delay);
//        return new CancellableDelegatingFuture<T>(future, nodeEngine, uuid, partitionId);
    }
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
@Override
public MapReduceTask<KeyIn, ValueIn, KeyOut, ValueOut> build( IMap<KeyIn, ValueIn> map )
{
    try
    {
        MapProxyImpl<KeyIn, ValueIn> proxy = (MapProxyImpl<KeyIn, ValueIn>) map;
        NodeEngine nodeEngine = hazelcastInstance.node.nodeEngine;
        return new IMapNodeMapReduceTaskImpl<KeyIn, ValueIn, KeyOut, ValueOut>( proxy.getName(), nodeEngine,
                                                                                hazelcastInstance );
    }
    catch ( Throwable t )
    {
        ExceptionUtil.rethrow( t );
    }
    return null;
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
Networking(NodeEngine nodeEngine, JobExecutionService jobExecutionService, int flowControlPeriodMs) {
    this.nodeEngine = (NodeEngineImpl) nodeEngine;
    this.logger = nodeEngine.getLogger(getClass());
    this.jobExecutionService = jobExecutionService;
    this.flowControlSender = nodeEngine.getExecutionService().scheduleWithRepetition(
            this::broadcastFlowControlPacket, 0, flowControlPeriodMs, MILLISECONDS);
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
public static byte[] createStreamPacketHeader(NodeEngine nodeEngine, long executionId,
                                              int destinationVertexId, int ordinal) {
    ObjectDataOutput out = createObjectDataOutput(nodeEngine);
    try {
        out.writeLong(executionId);
        out.writeInt(destinationVertexId);
        out.writeInt(ordinal);
        return out.toByteArray();
    } catch (IOException e) {
        throw sneakyThrow(e);
    }
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
public StoreSnapshotTasklet(SnapshotContext snapshotContext, long jobId, InboundEdgeStream inboundEdgeStream,
                            NodeEngine nodeEngine, String vertexName, boolean isHigherPrioritySource) {
    this.snapshotContext = snapshotContext;
    this.jobId = jobId;
    this.inboundEdgeStream = inboundEdgeStream;
    this.vertexName = vertexName;
    this.isHigherPrioritySource = isHigherPrioritySource;

    this.mapWriter = new AsyncMapWriter(nodeEngine);
    this.pendingSnapshotId = snapshotContext.lastSnapshotId() + 1;
    this.mapWriter.setMapName(currMapName());
    this.logger = nodeEngine.getLogger(StoreSnapshotTasklet.class + "." + vertexName + "#snapshot");
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
public ExecutionContext(NodeEngine nodeEngine, TaskletExecutionService execService,
                        long jobId, long executionId, Address coordinator, Set<Address> participants) {
    this.jobId = jobId;
    this.executionId = executionId;
    this.coordinator = coordinator;
    this.participants = new HashSet<>(participants);
    this.execService = execService;
    this.nodeEngine = nodeEngine;

    logger = nodeEngine.getLogger(getClass());
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
public SenderTasklet(InboundEdgeStream inboundEdgeStream, NodeEngine nodeEngine, Address destinationAddress,
                     long executionId, int destinationVertexId, int packetSizeLimit) {
    this.inboundEdgeStream = inboundEdgeStream;
    this.packetSizeLimit = packetSizeLimit;
    this.connection = getMemberConnection(nodeEngine, destinationAddress);
    this.outputBuffer = createObjectDataOutput(nodeEngine);
    uncheckRun(() -> outputBuffer.write(createStreamPacketHeader(
            nodeEngine, executionId, destinationVertexId, inboundEdgeStream.ordinal())));
    bufPosPastHeader = outputBuffer.position();
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
@Override
public void init(NodeEngine engine, Properties properties) {
    if (config == null) {
        throw new IllegalStateException("JetConfig is not initialized");
    }

    jetInstance = new JetInstanceImpl((HazelcastInstanceImpl) engine.getHazelcastInstance(), config);
    taskletExecutionService = new TaskletExecutionService(nodeEngine.getHazelcastInstance(),
            config.getInstanceConfig().getCooperativeThreadCount());

    snapshotRepository = new SnapshotRepository(jetInstance);
    jobRepository = new JobRepository(jetInstance, snapshotRepository);

    jobExecutionService = new JobExecutionService(nodeEngine, taskletExecutionService);
    jobCoordinationService = new JobCoordinationService(nodeEngine, config, jobRepository,
            jobExecutionService, snapshotRepository);
    networking = new Networking(engine, jobExecutionService, config.getInstanceConfig().getFlowControlPeriodMs());

    ClientEngineImpl clientEngine = engine.getService(ClientEngineImpl.SERVICE_NAME);
    ExceptionUtil.registerJetExceptions(clientEngine.getClientExceptionFactory());

    jobCoordinationService.init();

    JetBuildInfo jetBuildInfo = BuildInfoProvider.getBuildInfo().getJetBuildInfo();
    logger.info(String.format("Starting Jet %s (%s - %s)",
            jetBuildInfo.getVersion(), jetBuildInfo.getBuild(), jetBuildInfo.getRevision()));
    logger.info("Setting number of cooperative threads and default parallelism to "
            + config.getInstanceConfig().getCooperativeThreadCount());

    logger.info('\n' +
            "\to   o   o   o---o o---o o     o---o   o   o---o o-o-o        o o---o o-o-o\n" +
            "\t|   |  / \\     /  |     |     |      / \\  |       |          | |       |  \n" +
            "\to---o o---o   o   o-o   |     o     o---o o---o   |          | o-o     |  \n" +
            "\t|   | |   |  /    |     |     |     |   |     |   |      \\   | |       |  \n" +
            "\to   o o   o o---o o---o o---o o---o o   o o---o   o       o--o o---o   o   ");
    logger.info("Copyright (c) 2008-2017, Hazelcast, Inc. All Rights Reserved.");
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
@Nonnull
public static List<Address> getRemoteMembers(@Nonnull NodeEngine engine) {
    final Member localMember = engine.getLocalMember();
    return engine.getClusterService().getMembers().stream()
                 .filter(m -> !m.equals(localMember))
                 .map(Member::getAddress)
                 .collect(toList());
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
public AsyncMapWriter(NodeEngine nodeEngine) {
    this.partitionService = nodeEngine.getPartitionService();
    this.operationService = nodeEngine.getOperationService();
    this.mapService = nodeEngine.getService(MapService.SERVICE_NAME);
    this.outputBuffers = new MapEntries[partitionService.getPartitionCount()];
    this.serializationService = nodeEngine.getSerializationService();
    this.executionService = nodeEngine.getExecutionService();
    this.logger = nodeEngine.getLogger(AsyncMapWriter.class);
    JetService jetService = nodeEngine.getService(JetService.SERVICE_NAME);
    this.numConcurrentOps = jetService.numConcurrentPutAllOps();
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
@Override
public void run()
        throws Exception {

    NodeSequencerService sequencerService = getService();
    String sequencerName = getSequencerName();

    SequencerDefinition definition = sequencerService.destroySequencer(sequencerName, true);

    // Definition might be already destroyed concurrently
    if (definition == null) {
        return;
    }

    backupCount = definition.getBackupCount();

    NodeEngine nodeEngine = getNodeEngine();
    OperationService operationService = nodeEngine.getOperationService();

    DestroySequencerOperation operation = new DestroySequencerOperation(sequencerName);
    for (MemberImpl member : nodeEngine.getClusterService().getMemberImpls()) {
        if (!member.localMember() && !member.getAddress().equals(getCallerAddress())) {
            operationService.invokeOnTarget(SERVICE_NAME, operation, member.getAddress());
        }
    }

    ClientDestroySequencerNotification notification = new ClientDestroySequencerNotification(sequencerName);
    Collection<EventRegistration> registrations = sequencerService.findClientChannelRegistrations(sequencerName, null);
    nodeEngine.getEventService().publishEvent(SERVICE_NAME, registrations, notification, 1);
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
@Override
public void run()
        throws Exception {

    String sequencerName = getSequencerName();

    NodeSequencerService sequencerService = getService();
    SequencerDefinition definition = sequencerService.destroySequencer(sequencerName, true);

    // Definition might be already destroyed concurrently
    if (definition == null) {
        return;
    }

    backupCount = definition.getBackupCount();

    NodeEngine nodeEngine = getNodeEngine();

    OperationService operationService = nodeEngine.getOperationService();
    DestroySequencerOperation operation = new DestroySequencerOperation(sequencerName);
    for (MemberImpl member : nodeEngine.getClusterService().getMemberImpls()) {
        if (!member.localMember()) {
            operationService.invokeOnTarget(SERVICE_NAME, operation, member.getAddress());
        }
    }

    String clientUuid = getMessageChannel().getUuid();

    ClientDestroySequencerNotification notification = new ClientDestroySequencerNotification(sequencerName);
    Collection<EventRegistration> registrations = sequencerService.findClientChannelRegistrations(sequencerName, clientUuid);
    EventService eventService = nodeEngine.getEventService();
    eventService.publishEvent(SERVICE_NAME, registrations, notification, 1);
    eventService.deregisterAllListeners(SERVICE_NAME, sequencerName);
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
public SnowcastMessageTaskFactoryProvider(NodeEngine nodeEngine) {
    this.node = ((NodeEngineImpl) nodeEngine).getNode();
    register(SnowcastAttachLogicalNodeCodec.RequestParameters.TYPE, AttachLogicalNodeMessageTask::new);
    register(SnowcastCreateSequencerDefinitionCodec.RequestParameters.TYPE, CreateSequencerDefinitionMessageTask::new);
    register(SnowcastDestroySequencerDefinitionCodec.RequestParameters.TYPE, DestroySequencerDefinitionMessageTask::new);
    register(SnowcastDetachLogicalNodeCodec.RequestParameters.TYPE, DetachLogicalNodeMessageTask::new);
    register(SnowcastRegisterChannelCodec.RequestParameters.TYPE, RegisterChannelMessageTask::new);
    register(SnowcastRemoveChannelCodec.RequestParameters.TYPE, RemoveChannelMessageTask::new);
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
@Override
public RemoteServiceDescriptor[] createRemoteServiceDescriptors() {
    return new RemoteServiceDescriptor[]{new RemoteServiceDescriptor() {
        @Override
        public String getServiceName() {
            return SnowcastConstants.SERVICE_NAME;
        }

        @Override
        public RemoteService getService(NodeEngine nodeEngine) {
            return new NodeSequencerService();
        }
    }};
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
@Nonnull
private NodeSequencerService getSequencerService(@Nonnull NodeEngine nodeEngine) {
    return ExceptionUtils.execute(() -> {
        NodeSequencerService service = nodeEngine.getService(SnowcastConstants.SERVICE_NAME);
        printStartupMessage(false);
        return service;
    }, SERVICE_NOT_REGISTERED);
}
 

import com.hazelcast.spi.NodeEngine; //导入依赖的package包/类
@Nonnull
private NodeEngine getNodeEngine(@Nonnull HazelcastInstance hazelcastInstance) {
    return ExceptionUtils.execute(() -> {
        // Ugly hack due to lack in SPI
        //ACCESSIBILITY_HACK
        Field originalField = HazelcastInstanceProxy.class.getDeclaredField("original");
        originalField.setAccessible(true);
        HazelcastInstanceImpl impl = (HazelcastInstanceImpl) originalField.get(hazelcastInstance);
        return impl.node.getNodeEngine();
    }, RETRIEVE_NODE_ENGINE_FAILED);
}