<commit_before>/*
 * Copyright (c) 2008-2016, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.hazelcast.jet.impl.statemachine.applicationmaster.processors;

import com.hazelcast.core.IFunction;
import com.hazelcast.jet.api.application.ApplicationContext;
import com.hazelcast.jet.api.container.ContainerPayLoadProcessor;
import com.hazelcast.jet.api.container.DataChannel;
import com.hazelcast.jet.api.container.ProcessingContainer;
import com.hazelcast.jet.api.container.applicationmaster.ApplicationMaster;
import com.hazelcast.jet.api.processor.ContainerProcessorFactory;
import com.hazelcast.jet.impl.container.DefaultDataChannel;
import com.hazelcast.jet.impl.container.DefaultProcessingContainer;
import com.hazelcast.jet.impl.data.tuple.DefaultTupleFactory;
import com.hazelcast.jet.impl.statemachine.container.requests.ContainerStartRequest;
import com.hazelcast.jet.impl.util.JetUtil;
import com.hazelcast.jet.spi.config.JetApplicationConfig;
import com.hazelcast.jet.spi.dag.DAG;
import com.hazelcast.jet.spi.dag.Edge;
import com.hazelcast.jet.spi.dag.Vertex;
import com.hazelcast.jet.spi.data.tuple.TupleFactory;
import com.hazelcast.jet.spi.processor.ProcessorDescriptor;
import com.hazelcast.spi.NodeEngine;

import java.lang.reflect.Constructor;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static com.hazelcast.util.Preconditions.checkNotNull;

public class ExecutionPlanBuilderProcessor implements ContainerPayLoadProcessor<DAG> {
    private final NodeEngine nodeEngine;
    private final TupleFactory tupleFactory;
    private final ClassLoader applicationClassLoader;
    private final ApplicationMaster applicationMaster;
    private final ApplicationContext applicationContext;


    private final IFunction<Vertex, ProcessingContainer> containerProcessingCreator =
            new IFunction<Vertex, ProcessingContainer>() {
                @Override
                public ProcessingContainer apply(Vertex vertex) {
                    ProcessorDescriptor descriptor = vertex.getDescriptor();
                    String className = descriptor.getContainerProcessorFactoryClazz();
                    Object[] args = descriptor.getFactoryArgs();

                    ContainerProcessorFactory processorFactory = containerProcessorFactory(className, args);

                    ProcessingContainer container = new DefaultProcessingContainer(
                            vertex,
                            processorFactory,
                            nodeEngine,
                            applicationContext,
                            tupleFactory
                    );

                    applicationMaster.registerContainer(vertex, container);

                    return container;
                }
            };

    public ExecutionPlanBuilderProcessor(ApplicationMaster applicationMaster) {
        this.applicationMaster = applicationMaster;
        this.nodeEngine = applicationMaster.getNodeEngine();
        this.applicationContext = applicationMaster.getApplicationContext();
        this.applicationClassLoader = this.applicationContext.getLocalizationStorage().getClassLoader();
        this.tupleFactory = new DefaultTupleFactory();
    }

    @Override
    public void process(DAG dag) throws Exception {
        checkNotNull(dag);

        //Process dag and container's chain building
        Iterator<Vertex> iterator = dag.getTopologicalVertexIterator();

        Map<Vertex, ProcessingContainer> vertex2ContainerMap = new HashMap<Vertex, ProcessingContainer>(dag.getVertices().size());

        System.out.println("process1");

        List<ProcessingContainer> roots = new ArrayList<ProcessingContainer>();

        System.out.println("process2");

        while (iterator.hasNext()) {
            Vertex vertex = iterator.next();

            System.out.println("vertex=" + vertex.getName());

            List<Edge> edges = vertex.getInputEdges();
            ProcessingContainer next = this.containerProcessingCreator.apply(vertex);

            System.out.println("vertex=" + vertex.getName() + " out");

            vertex2ContainerMap.put(vertex, next);

            if (edges.size() == 0) {
                roots.add(next);
            } else {
                for (Edge edge : edges) {
                    join(vertex2ContainerMap.get(edge.getInputVertex()), edge, next);
                }
            }
        }
        System.out.println("process3");

        JetApplicationConfig jetApplicationConfig = this.applicationContext.getJetApplicationConfig();

        long secondsToAwait =
                jetApplicationConfig.getJetSecondsToAwait();

        System.out.println("process4");

        for (ProcessingContainer container : roots) {
            this.applicationMaster.addFollower(container);
        }

        System.out.println("process5");

        this.applicationMaster.deployNetworkEngine();

        System.out.println("process6");

        for (ProcessingContainer container : this.applicationMaster.containers()) {
            container.handleContainerRequest(new ContainerStartRequest()).get(secondsToAwait, TimeUnit.SECONDS);
        }

        System.out.println("process7");
    }

    private ProcessingContainer join(ProcessingContainer container, Edge edge, ProcessingContainer nextContainer) {
        linkWithChannel(container, nextContainer, edge);
        return nextContainer;
    }

    private void linkWithChannel(ProcessingContainer container,
                                 ProcessingContainer next,
                                 Edge edge
    ) {
        if ((container != null) && (next != null)) {
            DataChannel channel = new DefaultDataChannel(container, next, edge);

            container.addFollower(next);
            next.addPredecessor(container);

            container.addOutputChannel(channel);
            next.addInputChannel(channel);
        }
    }

    @SuppressWarnings("unchecked")
    private ContainerProcessorFactory containerProcessorFactory(String className, Object... args) {
        try {
            Class<ContainerProcessorFactory> clazz =
                    (Class<ContainerProcessorFactory>) Class.forName(
                            className,
                            true,
                            this.applicationClassLoader
                    );

            int i = 0;
            Class[] argsClasses = new Class[args.length];

            for (Object obj : args) {
                argsClasses[i++] = obj.getClass();
            }

            Constructor<ContainerProcessorFactory> resultConstructor = null;

            for (Constructor constructor : clazz.getConstructors()) {
                if (constructor.getParameterTypes().length == argsClasses.length) {
                    boolean valid = true;
                    for (int idx = 0; idx < argsClasses.length; idx++) {
                        if (!constructor.getParameterTypes()[idx].isAssignableFrom(argsClasses[idx])) {
                            valid = false;
                            break;
                        }
                    }

                    if (valid) {
                        resultConstructor = constructor;
                    }
                }
            }

            if (resultConstructor == null) {
                throw new IllegalStateException("No constructor with arguments"
                        + Arrays.toString(argsClasses)
                        + " className=" + className
                );
            }

            return resultConstructor.newInstance(args);
        } catch (Throwable e) {
            throw JetUtil.reThrow(e);
        }
    }
}<commit_msg>Processor should be created even with null parameters.<commit_after>/*
 * Copyright (c) 2008-2016, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.hazelcast.jet.impl.statemachine.applicationmaster.processors;

import com.hazelcast.core.IFunction;
import com.hazelcast.jet.api.application.ApplicationContext;
import com.hazelcast.jet.api.container.ContainerPayLoadProcessor;
import com.hazelcast.jet.api.container.DataChannel;
import com.hazelcast.jet.api.container.ProcessingContainer;
import com.hazelcast.jet.api.container.applicationmaster.ApplicationMaster;
import com.hazelcast.jet.api.processor.ContainerProcessorFactory;
import com.hazelcast.jet.impl.container.DefaultDataChannel;
import com.hazelcast.jet.impl.container.DefaultProcessingContainer;
import com.hazelcast.jet.impl.data.tuple.DefaultTupleFactory;
import com.hazelcast.jet.impl.statemachine.container.requests.ContainerStartRequest;
import com.hazelcast.jet.impl.util.JetUtil;
import com.hazelcast.jet.spi.config.JetApplicationConfig;
import com.hazelcast.jet.spi.dag.DAG;
import com.hazelcast.jet.spi.dag.Edge;
import com.hazelcast.jet.spi.dag.Vertex;
import com.hazelcast.jet.spi.data.tuple.TupleFactory;
import com.hazelcast.jet.spi.processor.ProcessorDescriptor;
import com.hazelcast.spi.NodeEngine;

import java.lang.reflect.Constructor;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static com.hazelcast.util.Preconditions.checkNotNull;

public class ExecutionPlanBuilderProcessor implements ContainerPayLoadProcessor<DAG> {
    private final NodeEngine nodeEngine;
    private final TupleFactory tupleFactory;
    private final ClassLoader applicationClassLoader;
    private final ApplicationMaster applicationMaster;
    private final ApplicationContext applicationContext;


    private final IFunction<Vertex, ProcessingContainer> containerProcessingCreator =
            new IFunction<Vertex, ProcessingContainer>() {
                @Override
                public ProcessingContainer apply(Vertex vertex) {
                    ProcessorDescriptor descriptor = vertex.getDescriptor();
                    String className = descriptor.getContainerProcessorFactoryClazz();
                    Object[] args = descriptor.getFactoryArgs();

                    ContainerProcessorFactory processorFactory = containerProcessorFactory(className, args);

                    ProcessingContainer container = new DefaultProcessingContainer(
                            vertex,
                            processorFactory,
                            nodeEngine,
                            applicationContext,
                            tupleFactory
                    );

                    applicationMaster.registerContainer(vertex, container);

                    return container;
                }
            };

    public ExecutionPlanBuilderProcessor(ApplicationMaster applicationMaster) {
        this.applicationMaster = applicationMaster;
        this.nodeEngine = applicationMaster.getNodeEngine();
        this.applicationContext = applicationMaster.getApplicationContext();
        this.applicationClassLoader = this.applicationContext.getLocalizationStorage().getClassLoader();
        this.tupleFactory = new DefaultTupleFactory();
    }

    @Override
    public void process(DAG dag) throws Exception {
        checkNotNull(dag);

        //Process dag and container's chain building
        Iterator<Vertex> iterator = dag.getTopologicalVertexIterator();

        Map<Vertex, ProcessingContainer> vertex2ContainerMap = new HashMap<Vertex, ProcessingContainer>(dag.getVertices().size());

        System.out.println("process1");

        List<ProcessingContainer> roots = new ArrayList<ProcessingContainer>();

        System.out.println("process2");

        while (iterator.hasNext()) {
            Vertex vertex = iterator.next();

            System.out.println("vertex=" + vertex.getName());

            List<Edge> edges = vertex.getInputEdges();
            ProcessingContainer next = this.containerProcessingCreator.apply(vertex);

            System.out.println("vertex=" + vertex.getName() + " out");

            vertex2ContainerMap.put(vertex, next);

            if (edges.size() == 0) {
                roots.add(next);
            } else {
                for (Edge edge : edges) {
                    join(vertex2ContainerMap.get(edge.getInputVertex()), edge, next);
                }
            }
        }
        System.out.println("process3");

        JetApplicationConfig jetApplicationConfig = this.applicationContext.getJetApplicationConfig();

        long secondsToAwait =
                jetApplicationConfig.getJetSecondsToAwait();

        System.out.println("process4");

        for (ProcessingContainer container : roots) {
            this.applicationMaster.addFollower(container);
        }

        System.out.println("process5");

        this.applicationMaster.deployNetworkEngine();

        System.out.println("process6");

        for (ProcessingContainer container : this.applicationMaster.containers()) {
            container.handleContainerRequest(new ContainerStartRequest()).get(secondsToAwait, TimeUnit.SECONDS);
        }

        System.out.println("process7");
    }

    private ProcessingContainer join(ProcessingContainer container, Edge edge, ProcessingContainer nextContainer) {
        linkWithChannel(container, nextContainer, edge);
        return nextContainer;
    }

    private void linkWithChannel(ProcessingContainer container,
                                 ProcessingContainer next,
                                 Edge edge
    ) {
        if ((container != null) && (next != null)) {
            DataChannel channel = new DefaultDataChannel(container, next, edge);

            container.addFollower(next);
            next.addPredecessor(container);

            container.addOutputChannel(channel);
            next.addInputChannel(channel);
        }
    }

    @SuppressWarnings("unchecked")
    private ContainerProcessorFactory containerProcessorFactory(String className, Object... args) {
        try {
            Class<ContainerProcessorFactory> clazz =
                    (Class<ContainerProcessorFactory>) Class.forName(
                            className,
                            true,
                            this.applicationClassLoader
                    );

            int i = 0;
            Class[] argsClasses = new Class[args.length];

            for (Object obj : args) {
                if (obj != null) {
                    argsClasses[i++] = obj.getClass();
                }
            }

            Constructor<ContainerProcessorFactory> resultConstructor = null;

            for (Constructor constructor : clazz.getConstructors()) {
                if (constructor.getParameterTypes().length == argsClasses.length) {
                    boolean valid = true;
                    for (int idx = 0; idx < argsClasses.length; idx++) {
                        Class argsClass = argsClasses[idx];
                        if (argsClass != null && !constructor.getParameterTypes()[idx].isAssignableFrom(argsClass)) {
                            valid = false;
                            break;
                        }
                    }

                    if (valid) {
                        resultConstructor = constructor;
                    }
                }
            }

            if (resultConstructor == null) {
                throw new IllegalStateException("No constructor with arguments"
                        + Arrays.toString(argsClasses)
                        + " className=" + className
                );
            }

            return resultConstructor.newInstance(args);
        } catch (Throwable e) {
            throw JetUtil.reThrow(e);
        }
    }
}
{'fraction_non_alphanumeric': 0.08651565464895636, 'fraction_numerical': 0.0029648956356736244, 'mean_word_length': 2.3270862102978893, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4830105', 'n_tokens_mistral': 4056, 'n_tokens_neox': 3853, 'n_words': 853}