/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.flink.graph.partition.centric.utils;

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.Histogram;
import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.graph.Graph;
import org.apache.flink.graph.library.ConnectedComponents;
import org.apache.flink.graph.partition.centric.PCConnectedComponents;
import org.apache.flink.graph.partition.centric.PartitionCentricConfiguration;
import org.apache.flink.graph.partition.centric.PartitionCentricIteration;
import org.apache.flink.types.NullValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * Utility class to run connected component algorithm on different graph
 */
public class GraphCCRunner {
    private final static Logger LOG = LoggerFactory.getLogger(GraphCCRunner.class);

    public static <K, EV> void detectComponentVC(
            ExecutionEnvironment environment,
            Graph<K, Long, EV> graph,
            String vertexCentricOutput) throws Exception {

        LOG.debug("Start warming up JVM");
        Telemetry.dummyPCConnectedComponent(environment);
        LOG.debug("JVM warm up finished");

        JobExecutionResult result;
        Map<String, String> fields = new HashMap<>();

        environment.startNewSession();
        ConnectedComponents<K, EV> vcAlgo = new ConnectedComponents<>(Integer.MAX_VALUE);
        vcAlgo.run(graph).writeAsCsv(vertexCentricOutput, FileSystem.WriteMode.OVERWRITE);
        result = environment.execute();
        fields.clear();
        Telemetry.printTelemetry("Vertex centric", result, fields);
    }

    public static <K, EV> void detectComponentPC(ExecutionEnvironment environment,
                                         Graph<K, Long, EV> graph,
                                         String partitionCentricOutput) throws Exception {
        // Run some dummy computation to warmup the jvm
        LOG.debug("Start warming up JVM");
        Telemetry.dummyVCConnectedComponent(environment);
        LOG.debug("JVM warm up finished");

        JobExecutionResult result;
        PartitionCentricConfiguration configuration = new PartitionCentricConfiguration();
        configuration.registerAccumulator(PCConnectedComponents.MESSAGE_SENT_CTR, new LongCounter());
        configuration.registerAccumulator(PCConnectedComponents.MESSAGE_SENT_ITER_CTR, new Histogram());
        configuration.registerAccumulator(PCConnectedComponents.ITER_CTR, new LongCounter());
        configuration.registerAccumulator(PCConnectedComponents.ACTIVE_VER_ITER_CTR, new Histogram());
        configuration.registerAccumulator(PartitionCentricIteration.ITER_TIMER, new IterationTimer());

        environment.startNewSession();
        PCConnectedComponents<K, EV> algo = new PCConnectedComponents<>(
                Integer.MAX_VALUE, configuration);
        algo.run(graph).writeAsCsv(partitionCentricOutput, FileSystem.WriteMode.OVERWRITE);
        result = environment.execute();
        Map<String, String> fields = new HashMap<>();
        fields.put(PCConnectedComponents.MESSAGE_SENT_CTR, "Total messages sent");
        fields.put(PCConnectedComponents.MESSAGE_SENT_ITER_CTR, "Messages sent");
        fields.put(PCConnectedComponents.ITER_CTR, "Iteration count");
        fields.put(PCConnectedComponents.ACTIVE_VER_ITER_CTR, "Active vertices");
        fields.put(PartitionCentricIteration.ITER_TIMER, "Elapse time");
        Telemetry.printTelemetry("Partition centric", result, fields);
    }
}

@@ -2057,17 +2057,17 @@
 ry.dummy
-P
+V
 CConnect
@@ -2940,33 +2940,33 @@
  Telemetry.dummy
-V
+P
 CConnectedCompon

Fix wrong warm up code for ConnectedComponent