import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
    if (args.length == 0) {
        System.out.println("Usage: hdfs-to-map <name> <input path> <parallelism>");
        return;
    }

    String name = args[0];
    String inputPath = args[1];
    int parallelism = Integer.parseInt(args[2]);

    JetInstance client = Jet.newJetClient();
    IStreamMap<Long, String> map = client.getMap(name);
    map.clear();

    try {
        long begin = System.currentTimeMillis();
        fillMap(client, name, inputPath, parallelism);
        long elapsed = System.currentTimeMillis() - begin;
        System.out.println("Time=" + elapsed);
    } finally {
        client.shutdown();
    }
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
private static void fillMap(JetInstance client, String name, String inputPath, int parallelism) throws Exception {
    DAG dag = new DAG();
    JobConf conf = new JobConf();
    conf.setInputFormat(TextInputFormat.class);
    TextInputFormat.addInputPath(conf, new Path(inputPath));


    Vertex reader = dag.newVertex("reader", readHdfsP(conf, Util::entry));
    Vertex mapper = dag.newVertex("mapper",
            mapP((Map.Entry<LongWritable, Text> e) -> entry(e.getKey().get(), e.getValue().toString())));
    Vertex writer = dag.newVertex("writer", writeMapP(name));

    reader.localParallelism(parallelism);
    mapper.localParallelism(parallelism);
    writer.localParallelism(parallelism);

    dag.edge(between(reader, mapper));
    dag.edge(between(mapper, writer));


    JobConfig jobConfig = new JobConfig();
    jobConfig.addClass(HdfsToMap.class);

    client.newJob(dag, jobConfig).join();
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
@Test
public void test_Jar_Distribution() throws Throwable {
    createCluster();

    DAG dag = new DAG();
    dag.newVertex("create and print person", LoadPersonIsolated::new);


    JetInstance jetInstance = getJetInstance();
    JobConfig jobConfig = new JobConfig();
    jobConfig.addJar(this.getClass().getResource("/sample-pojo-1.0-person.jar"));
    jobConfig.addJar(this.getClass().getResource("/sample-pojo-1.0-deployment.jar"));
    jobConfig.addClass(AbstractDeploymentTest.class);

    executeAndPeel(jetInstance.newJob(dag, jobConfig));
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
@Before
public void setUp() {
    JetConfig config = new JetConfig();

    EventJournalConfig journalConfig = new EventJournalConfig()
            .setMapName("*")
            .setCapacity(JOURNAL_CAPACITY)
            .setEnabled(true);

    config.getHazelcastConfig().setProperty(PARTITION_COUNT.getName(), String.valueOf(NUM_PARTITIONS));
    config.getHazelcastConfig().addEventJournalConfig(journalConfig);
    JetInstance instance = this.createJetMember(config);

    map = (MapProxyImpl<Integer, Integer>) instance.getHazelcastInstance().<Integer, Integer>getMap("test");
    List<Integer> allPartitions = IntStream.range(0, NUM_PARTITIONS).boxed().collect(toList());

    supplier = () -> new StreamEventJournalP<>(map, allPartitions, e -> true,
            EventJournalMapEvent::getNewValue, START_FROM_OLDEST, false,
            wmGenParams(Integer::intValue, withFixedLag(0), suppressAll(), -1));
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
@Test
public void when_writeBufferedJobFailed_then_bufferDisposed() throws Exception {
    JetInstance instance = createJetMember();
    try {
        DAG dag = new DAG();
        Vertex source = dag.newVertex("source", StuckForeverSourceP::new);
        Vertex sink = dag.newVertex("sink", getLoggingBufferedWriter()).localParallelism(1);

        dag.edge(Edge.between(source, sink));

        Job job = instance.newJob(dag);
        // wait for the job to initialize
        Thread.sleep(5000);
        job.cancel();

        assertTrueEventually(() -> assertTrue("No \"dispose\", only: " + events, events.contains("dispose")), 60);
        System.out.println(events);
    } finally {
        instance.shutdown();
    }
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
@Test
public void test_serializationFromNodeToClient() {
    // create one member and one client
    createJetMember();
    JetInstance client = createJetClient();

    RuntimeException exc = new RuntimeException("myException");
    try {
        DAG dag = new DAG();
        dag.newVertex("source", () -> new ProcessorThatFailsInComplete(exc)).localParallelism(1);
        client.newJob(dag).join();
    } catch (Exception caught) {
        assertThat(caught.toString(), containsString(exc.toString()));
        TestUtil.assertExceptionInCauses(exc, caught);
    } finally {
        shutdownFactory();
    }
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
@Test
public void test_serializationOnNode() {
    JetTestInstanceFactory factory = new JetTestInstanceFactory();
    // create one member and one client
    JetInstance client = factory.newMember();

    RuntimeException exc = new RuntimeException("myException");
    try {
        DAG dag = new DAG();
        dag.newVertex("source", () -> new ProcessorThatFailsInComplete(exc)).localParallelism(1);
        client.newJob(dag).join();
    } catch (Exception caught) {
        assertThat(caught.toString(), containsString(exc.toString()));
        TestUtil.assertExceptionInCauses(exc, caught);
    } finally {
        factory.terminateAll();
    }
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
private void testJobSubmissionTimeWhenJobIsRunning(JetInstance instance) throws InterruptedException {
    // Given
    DAG dag = new DAG().vertex(new Vertex("test", new MockPS(StuckProcessor::new, NODE_COUNT)));
    JobConfig config = new JobConfig();
    String jobName = "job1";
    config.setName(jobName);

    // When
    Job job = instance1.newJob(dag, config);
    StuckProcessor.executionStarted.await();
    Job trackedJob = instance.getJob("job1");

    // Then
    assertNotNull(trackedJob);
    assertNotEquals(0, job.getSubmissionTime());
    assertNotEquals(0, trackedJob.getSubmissionTime());
    StuckProcessor.proceedLatch.countDown();
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
@Before
public void setup() {
    factory = new JetTestInstanceFactory();

    JetConfig config = new JetConfig();
    config.getInstanceConfig().setCooperativeThreadCount(LOCAL_PARALLELISM);

    // force snapshots to fail by adding a failing map store configuration for snapshot data maps
    MapConfig mapConfig = new MapConfig(SnapshotRepository.SNAPSHOT_DATA_NAME_PREFIX + '*');
    MapStoreConfig mapStoreConfig = mapConfig.getMapStoreConfig();
    mapStoreConfig.setEnabled(true);
    mapStoreConfig.setImplementation(new FailingMapStore());
    config.getHazelcastConfig().addMapConfig(mapConfig);

    JetInstance[] instances = factory.newMembers(config, 2);
    instance1 = instances[0];
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
@Test
public void when_jobCancelledOnSingleNode_then_terminatedEventually() {
    // Given
    JetInstance instance = newInstance();

    DAG dag = new DAG();
    dag.newVertex("slow", StuckSource::new);

    Job job = instance.newJob(dag);
    assertExecutionStarted();

    // When
    job.cancel();

    // Then
    assertExecutionTerminated();
    expectedException.expect(CancellationException.class);
    job.join();
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
@Test
public void when_jobCancelledOnMultipleNodes_then_terminatedEventually() {
    // Given
    newInstance();
    JetInstance instance = newInstance();

    DAG dag = new DAG();
    dag.newVertex("slow", StuckSource::new);

    Job job = instance.newJob(dag);
    assertExecutionStarted();

    // When
    job.cancel();

    // Then
    assertExecutionTerminated();
    expectedException.expect(CancellationException.class);
    job.join();
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
@Test
public void when_jobCancelled_then_jobStatusIsSetEventually() {
    // Given
    JetInstance instance = newInstance();

    DAG dag = new DAG();
    dag.newVertex("slow", StuckSource::new);

    Job job = instance.newJob(dag);
    assertExecutionStarted();

    // When
    job.cancel();

    // Then
    assertTrueEventually(() -> assertEquals(JobStatus.COMPLETED, job.getStatus()), 3);
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
@Test
public void when_jobCancelledFromClient_then_terminatedEventually() {
    // Given
    newInstance();
    newInstance();
    JetInstance client = factory.newClient();

    DAG dag = new DAG();
    dag.newVertex("slow", StuckSource::new);

    Job job = client.newJob(dag);
    assertExecutionStarted();

    // When
    job.cancel();

    // Then
    assertExecutionTerminated();
    expectedException.expect(CancellationException.class);
    job.join();
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
@Test
public void when_jobCancelledFromClient_then_jobStatusIsSetEventually() {
    // Given
    newInstance();
    newInstance();
    JetInstance client = factory.newClient();

    DAG dag = new DAG();
    dag.newVertex("slow", StuckSource::new);

    Job job = client.newJob(dag);
    assertExecutionStarted();

    // When
    job.cancel();

    // Then
    assertTrueEventually(() -> assertEquals(JobStatus.COMPLETED, job.getStatus()), 3);
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
@Test
public void when_jobCancelled_then_trackedJobsGetNotified() {
    // Given
    JetInstance instance1 = newInstance();
    JetInstance instance2 = newInstance();

    DAG dag = new DAG();
    dag.newVertex("slow", StuckSource::new);

    Job job = instance1.newJob(dag);
    assertExecutionStarted();

    // When
    job.cancel();

    // Then
    assertExecutionTerminated();
    expectedException.expect(CancellationException.class);
    Job tracked = instance2.getJobs().iterator().next();
    tracked.join();
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
@Test
public void when_jobCancelled_then_jobStatusIsSetDuringCancellation() {
    // Given
    JetInstance instance1 = newInstance();
    JetInstance instance2 = newInstance();
    dropOperationsBetween(instance1.getHazelcastInstance(), instance2.getHazelcastInstance(),
            JetInitDataSerializerHook.FACTORY_ID, singletonList(JetInitDataSerializerHook.COMPLETE_EXECUTION_OP));

    DAG dag = new DAG();
    dag.newVertex("slow", StuckSource::new);

    Job job = instance1.newJob(dag);
    assertExecutionStarted();

    // When
    job.cancel();

    // Then
    assertTrueEventually(() -> assertEquals(JobStatus.COMPLETING, job.getStatus()), 3);

    resetPacketFiltersFrom(instance1.getHazelcastInstance());

    assertTrueEventually(() -> assertEquals(JobStatus.COMPLETED, job.getStatus()), 3);
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
private void testGetJobByNameWhenMultipleJobsAreRunning(JetInstance instance) throws InterruptedException {
    // Given
    DAG dag = new DAG().vertex(new Vertex("test", new MockPS(StuckProcessor::new, NODE_COUNT * 2)));
    JobConfig config = new JobConfig();
    String jobName = "job1";
    config.setName(jobName);

    // When
    Job job1 = instance1.newJob(dag, config);
    sleepAtLeastMillis(1);
    Job job2 = instance1.newJob(dag, config);
    StuckProcessor.executionStarted.await();

    // Then
    Job trackedJob = instance.getJob(jobName);

    assertNotNull(trackedJob);
    assertEquals(jobName, trackedJob.getName());
    assertNotEquals(job1.getId(), trackedJob.getId());
    assertEquals(job2.getId(), trackedJob.getId());
    assertTrueEventually(() -> assertEquals(RUNNING, trackedJob.getStatus()));

    StuckProcessor.proceedLatch.countDown();
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
/**
 * Starts a new {@code JetInstance} which is only able to communicate
 * with members on one of the two brains.
 * @param firstSubCluster jet instances in the first sub cluster
 * @param secondSubCluster jet instances in the first sub cluster
 * @param createOnFirstSubCluster if true, new instance is created on the first sub cluster.
 * @return a HazelcastInstance whose {@code MockJoiner} has blacklisted the other brain's
 *         members and its connection manager blocks connections to other brain's members
 * @see TestHazelcastInstanceFactory#newHazelcastInstance(Address, com.hazelcast.config.Config, Address[])
 */
protected final JetInstance createHazelcastInstanceInBrain(JetInstance[] firstSubCluster,
                                                           JetInstance[] secondSubCluster,
                                                           boolean createOnFirstSubCluster) {
    Address newMemberAddress = nextAddress();
    JetInstance[] instancesToBlock = createOnFirstSubCluster ? secondSubCluster : firstSubCluster;

    List<Address> addressesToBlock = new ArrayList<>(instancesToBlock.length);
    for (JetInstance anInstancesToBlock : instancesToBlock) {
        if (isInstanceActive(anInstancesToBlock)) {
            addressesToBlock.add(getAddress(anInstancesToBlock.getHazelcastInstance()));
            // block communication from these instances to the new address

            FirewallingConnectionManager connectionManager = getFireWalledConnectionManager(
                    anInstancesToBlock.getHazelcastInstance());
            connectionManager.blockNewConnection(newMemberAddress);
            connectionManager.closeActiveConnection(newMemberAddress);
        }
    }
    // indicate we need to unblacklist addresses from joiner when split-brain will be healed
    unblacklistHint = true;
    // create a new Hazelcast instance which has blocked addresses blacklisted in its joiner
    return createJetMember(createConfig(), addressesToBlock.toArray(new Address[addressesToBlock.size()]));
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
private void applyOnBrains(JetInstance[] instances, int firstSubClusterSize,
                           BiConsumer<HazelcastInstance, HazelcastInstance> action) {
    for (int i = 0; i < firstSubClusterSize; i++) {
        JetInstance isolatedInstance = instances[i];
        // do not take into account instances which have been shutdown
        if (!isInstanceActive(isolatedInstance)) {
            continue;
        }
        for (int j = firstSubClusterSize; j < instances.length; j++) {
            JetInstance currentInstance = instances[j];
            if (!isInstanceActive(currentInstance)) {
                continue;
            }
            action.accept(isolatedInstance.getHazelcastInstance(), currentInstance.getHazelcastInstance());
        }
    }
}
 

import com.hazelcast.jet.JetInstance; //导入依赖的package包/类
@Test
public void when_newMemberJoinsToCluster_then_jobQuorumSizeIsUpdated() {
    int clusterSize = 3;
    JetConfig jetConfig = new JetConfig();
    JetInstance[] instances = new JetInstance[clusterSize];
    for (int i = 0; i < clusterSize; i++) {
        instances[i] = createJetMember(jetConfig);
    }

    StuckProcessor.executionStarted = new CountDownLatch(clusterSize * PARALLELISM);
    MockPS processorSupplier = new MockPS(StuckProcessor::new, clusterSize);
    DAG dag = new DAG().vertex(new Vertex("test", processorSupplier));
    Job job = instances[0].newJob(dag, new JobConfig().setSplitBrainProtection(true));
    assertOpenEventually(StuckProcessor.executionStarted);

    createJetMember(jetConfig);

    assertTrueEventually(() -> {
        JobRepository jobRepository = getJetService(instances[0]).getJobRepository();
        JobRecord jobRecord = jobRepository.getJobRecord(job.getId());
        assertEquals(3, jobRecord.getQuorumSize());
    });

    StuckProcessor.proceedLatch.countDown();
}