text:
package com.tlswe.awsmock.cloudwatch.control;

import java.io.IOException;
import java.util.GregorianCalendar;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

import javax.servlet.http.HttpServletResponse;
import javax.xml.datatype.DatatypeConfigurationException;
import javax.xml.datatype.DatatypeFactory;
import javax.xml.datatype.XMLGregorianCalendar;

import org.apache.commons.lang3.StringEscapeUtils;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.Seconds;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.tlswe.awsmock.cloudwatch.cxf_generated.Datapoint;
import com.tlswe.awsmock.cloudwatch.cxf_generated.Datapoints;
import com.tlswe.awsmock.cloudwatch.cxf_generated.DescribeAlarmsResponse;
import com.tlswe.awsmock.cloudwatch.cxf_generated.DescribeAlarmsResult;
import com.tlswe.awsmock.cloudwatch.cxf_generated.GetMetricStatisticsResponse;
import com.tlswe.awsmock.cloudwatch.cxf_generated.GetMetricStatisticsResult;
import com.tlswe.awsmock.cloudwatch.cxf_generated.MetricAlarm;
import com.tlswe.awsmock.cloudwatch.cxf_generated.MetricAlarms;
import com.tlswe.awsmock.cloudwatch.cxf_generated.ResponseMetadata;
import com.tlswe.awsmock.cloudwatch.cxf_generated.StandardUnit;
import com.tlswe.awsmock.cloudwatch.util.JAXBUtilCW;
import com.tlswe.awsmock.common.exception.AwsMockException;
import com.tlswe.awsmock.common.util.Constants;
import com.tlswe.awsmock.common.util.PropertiesUtils;
import com.tlswe.awsmock.common.util.TemplateUtils;
import com.tlswe.awsmock.ec2.cxf_generated.PlacementResponseType;
import com.tlswe.awsmock.ec2.exception.BadEc2RequestException;

/**
 * Class that handlers requests of AWS CloudWatch Query API for monitoring the mock ec2
 * instances. This class works between All object of mock CloudWatch.
 *
 * @author Davinder
 *
 */
public final class MockCloudWatchQueryHandler {

    /**
     * Singleton instance of MockCloudWatchQueryHandler.
     */
    private static MockCloudWatchQueryHandler singletonMockCloudWatchQueryHandler = null;

    /**
     * Log writer for this class.
     */
    private final Logger log = LoggerFactory.getLogger(MockCloudWatchQueryHandler.class);

    /**
     * The xml template filename for error response body.
     */
    private static final String ERROR_RESPONSE_TEMPLATE = "error.xml.ftl";

    /**
     * The xml template filename for blank response body.
     */
    private static final String BLANK_RESPONSE_TEMPLATE = "blank.xml.ftl";

    /**
     * Description for the link to AWS QUERY API reference.
     */
    private static final String REF_CLOUDWATCH_QUERY_API_DESC = "See http://docs.aws.amazon.com/AmazonCloudWatch/"
            + "latest/monitoring/WhatIsCloudWatch.html for building a valid query.";

    /**
     * Default placement for this aws-mock, defined in aws-mock.properties (or if not overridden, as defined in
     * aws-mock-default.properties).
     */
    private static final PlacementResponseType DEFAULT_MOCK_PLACEMENT = new PlacementResponseType();

    /**
     * Predefined AMIs, as properties of predefined.mock.ami.X in
     * aws-mock.properties (or if not overridden, as defined in
     * aws-mock-default.properties). We use {@link TreeSet} here so that those
     * AMIs are loaded and displayed (described) in the same order the are
     * defined in the .properties file.
     */
    private static final Set<String> MOCK_AMIS = new TreeSet<String>();

    /**
     * A common random generator.
     */
    private static Random random = new Random();

    /**
     * The chars used to generate tokens (those tokens in describeInstances
     * req/resp pagination).
     */
    private static final String TOKEN_DICT = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";

    /**
     * Token prefix length.
     *
     * AWS's token is 276 bytes in length (19 fixed + 240 generated per response
     * + 17 fixed), we just mock that way.
     */
    protected static final int TOKEN_PREFIX_LEN = 19;

    /**
     * Token suffix length.
     */
    protected static final int TOKEN_SUFFIX_LEN = 17;

    /**
     * Token mid-string length.
     */
    protected static final int TOKEN_MIDDLE_LEN = 240;

    /**
     * The prefix string, which would be determined on app startup.
     */
    protected static final String TOKEN_PREFIX;

    /**
     * The suffix string, which would be determined on app startup.
     */
    protected static final String TOKEN_SUFFIX;

    /**
     * Default page size for pagination in describeInstance response.
     */
    protected static final int MAX_RESULTS_DEFAULT = 1000;

    static {
        DEFAULT_MOCK_PLACEMENT.setAvailabilityZone(PropertiesUtils.getProperty(Constants.PROP_NAME_EC2_PLACEMENT));
        MOCK_AMIS.addAll(PropertiesUtils.getPropertiesByPrefix("predefined.mock.ami."));

        /*
         * We determine the token's prefix and suffix at the start of webapp and
         * don't change them.
         */
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < TOKEN_PREFIX_LEN; i++) {
            sb.append(TOKEN_DICT.charAt(random.nextInt(TOKEN_DICT.length())));
        }
        TOKEN_PREFIX = sb.toString();

        sb = new StringBuilder();
        for (int i = 0; i < TOKEN_SUFFIX_LEN; i++) {
            sb.append(TOKEN_DICT.charAt(random.nextInt(TOKEN_DICT.length())));
        }
        TOKEN_SUFFIX = sb.toString();
    }

    /**
     * Constructor of MockEC2QueryHandler is made private and only called once
     * by {@link #getInstance()}.
     */
    private MockCloudWatchQueryHandler() {
    }

    /**
     *
     * @return singleton instance of {@link MockCloudWatchQueryHandler}
     */
    public static MockCloudWatchQueryHandler getInstance() {
        if (null == singletonMockCloudWatchQueryHandler) {
            // "double lock lazy loading" for singleton instance loading on
            // first time usage
            synchronized (MockCloudWatchQueryHandler.class) {
                if (null == singletonMockCloudWatchQueryHandler) {
                    singletonMockCloudWatchQueryHandler = new MockCloudWatchQueryHandler();
                }
            }
        }
        return singletonMockCloudWatchQueryHandler;
    }

    /**
     * Hub method for parsing query prarmeters and generate and write xml
     * response.
     *
     * @param queryParams
     *            map of query parameters from http request, which is from
     *            standard AWS Query API
     * @param requestHeaders
     *            map of Headers from http request, which is from standard AWS Query API
     * @param response
     *            http servlet response to handle with
     * @throws IOException
     *             in case of failure of getting response's writer
     *
     */
     public void handle(final Map<String, String[]> queryParams,
                       final Map<String, String> requestHeaders, final HttpServletResponse response)
            throws IOException {
        if (null == response) {
            // do nothing in case null is passed in
            return;
        }

        String responseXml = null;

        if (null != requestHeaders && requestHeaders.size() > 0) {
            if (requestHeaders.containsKey("region")
                && !DEFAULT_MOCK_PLACEMENT.getAvailabilityZone().equals(requestHeaders.get("region").toLowerCase())) {
                response.getWriter().write(getBlankResponseXml());
                response.getWriter().flush();
                response.setStatus(HttpServletResponse.SC_OK);
                return;
            }
        }

        if (null == queryParams || queryParams.size() == 0) {
            // no params found at all - write an error xml response
            response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
            responseXml = getXmlError("InvalidQuery",
                    "No parameter in query at all! " + REF_CLOUDWATCH_QUERY_API_DESC);
        } else {
            // parse the parameters in query
            String[] versionParamValues = queryParams.get("Version");

            if (null == versionParamValues || versionParamValues.length != 1) {
                // no version param found - write an error xml response
                response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
                responseXml = getXmlError("InvalidQuery",
                        "There should be a parameter of 'Version' provided in the query! "
                                + REF_CLOUDWATCH_QUERY_API_DESC);
            } else {

                String version = versionParamValues[0];

                String[] actions = queryParams.get("Action");

                if (null == actions || actions.length != 1) {
                    // no action found - write response for error
                    response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
                    responseXml = getXmlError("InvalidQuery",
                            "There should be a parameter of 'Action' provided in the query! "
                                    + REF_CLOUDWATCH_QUERY_API_DESC);
                } else {

                    String action = actions[0];

                    try {

                        response.setStatus(HttpServletResponse.SC_OK);

                        if ("GetMetricStatistics".equals(action)) {

                            DateTimeZone zone = DateTimeZone.UTC;
                            DateTime startTime = new DateTime(queryParams.get("StartTime")[0],
                                    zone);
                            DateTime endTime = new DateTime(queryParams.get("EndTime")[0], zone);

                            String[] statistics = new String[2];
                            if (queryParams.containsKey("Statistics.member.1")) {
                                statistics[0] = queryParams.get("Statistics.member.1")[0];
                            }

                            if (queryParams.containsKey("Statistics.member.2")) {
                                statistics[1] = queryParams.get("Statistics.member.2")[0];
                            }

                            String metricName = queryParams.get("MetricName")[0];
                            int period = Integer.parseInt(queryParams.get("Period")[0]);
                            responseXml = JAXBUtilCW.marshall(
                                    getMetricStatistics(statistics, startTime, endTime, period,
                                            metricName),
                                    "GetMetricStatistics", version);
                        } else  if ("DescribeAlarms".equals(action)) {
                                responseXml = JAXBUtilCW.marshall(
                                        describeAlarms(),
                                        "DescribeAlarmsResponse", version);
                        } else {
                            // unsupported/unimplemented action - write an
                            // error
                            // response
                            response.setStatus(HttpServletResponse.SC_NOT_IMPLEMENTED);
                            String allImplementedActions = "runInstances|stopInstances|startInstances|"
                                    + "terminateInstances|describeInstances|describeImages";
                            responseXml = getXmlError("NotImplementedAction",
                                    "Action '" + action
                                            + "' has not been implemented yet in aws-mock. "
                                            + "For now we only support actions as following: "
                                            + allImplementedActions);
                        }
                    } catch (BadEc2RequestException e) {
                        response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
                        responseXml = getXmlError("InvalidQuery",
                                "invalid request for '" + action + "'. " + e.getMessage()
                                        + REF_CLOUDWATCH_QUERY_API_DESC);
                    } catch (AwsMockException e) {
                        log.error("server error occured while processing '{}' request. {}", action,
                                e.getMessage());
                        response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
                        responseXml = getXmlError("InternalError", e.getMessage());
                    } catch (Exception e) {
                        log.error("server error occured while processing '{}' request. {}", action,
                                e.getMessage());
                        response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
                        if (e.getMessage() != null) {
                             responseXml = getXmlError("InternalError", e.getMessage());
                        } else {
                             responseXml = "InternalError";
                        }
                    }

                }
            }

        }

        // System.out.println(responseXml);
        response.getWriter().write(responseXml);
        response.getWriter().flush();

    }

    /**
     * Handles "GetMetricStatistics" request, as simple as without any filters
     * to use.
     *
     * @param statistics
     *            Metric statistics.
     * @param startTime
     *            Metric statistics start time.
     * @param endTime
     *            Metric statistics end time.
     * @param period
     *            Metric collection period.
     * @param metricName
     *            Metric Name.
     * @return a GetMetricStatisticsResult for metricName.
     */
    private GetMetricStatisticsResponse getMetricStatistics(final String[] statistics,
            final DateTime startTime,
            final DateTime endTime, final int period, final String metricName) {
        GetMetricStatisticsResponse ret = new GetMetricStatisticsResponse();
        GetMetricStatisticsResult result = new GetMetricStatisticsResult();
        Datapoints dataPoints = new Datapoints();

        int dataPointsCount = Seconds.secondsBetween(startTime, endTime).getSeconds() / period;
        DateTime newDate = startTime;
        for (int counterDp = 0; counterDp < dataPointsCount; counterDp++) {
            Datapoint dp = new Datapoint();
            DateTime timeStamp = newDate.plusSeconds(period);
            XMLGregorianCalendar timeStampXml = toXMLGregorianCalendar(timeStamp);
            dp.setTimestamp(timeStampXml);
            dp.setAverage(getMetricAverageValue(metricName));
            dp.setSampleCount(getMetricSampleCountValue(metricName));
            dp.setUnit(getMetricUnit(metricName));
            dataPoints.getMember().add(dp);
            newDate = timeStamp;
        }

        result.setDatapoints(dataPoints);
        result.setLabel(metricName);
        ret.setGetMetricStatisticsResult(result);
        ResponseMetadata responseMetadata = new ResponseMetadata();
        responseMetadata.setRequestId(UUID.randomUUID().toString());
        ret.setResponseMetadata(responseMetadata);
        return ret;
    }

    /**
     * Handles "describeAlarms" request.
     *
     * @return a GetMetricStatisticsResult for metricName.
     */
    private DescribeAlarmsResponse describeAlarms() {
        DescribeAlarmsResponse ret = new DescribeAlarmsResponse();
        DescribeAlarmsResult result = new DescribeAlarmsResult();
        MetricAlarms metricAlarms = new MetricAlarms();

        MetricAlarm metricAlarm = new MetricAlarm();
        metricAlarm.setAlarmDescription("CPU usage exceeds 70 percent");
        metricAlarm.setAlarmName("AwsMockAlarmCPU");
        metricAlarms.getMember().add(metricAlarm);

        metricAlarm = new MetricAlarm();
        metricAlarm.setAlarmDescription("Memory usage exceeds 80 percent");
        metricAlarm.setAlarmName("AwsMockAlarmMemory");
        metricAlarms.getMember().add(metricAlarm);

        result.setMetricAlarms(metricAlarms);

        ret.setDescribeAlarmsResult(result);
        ResponseMetadata responseMetadata = new ResponseMetadata();
        responseMetadata.setRequestId(UUID.randomUUID().toString());
        ret.setResponseMetadata(responseMetadata);
        return ret;
    }

    /**
     * Converts java.util.Date to javax.xml.datatype.XMLGregorianCalendar.
     *
     * @param date
     *            Date parameter.
     * @return a XMLGregorianCalendar for date.
     */
    private XMLGregorianCalendar toXMLGregorianCalendar(final DateTime date) {
        GregorianCalendar gCalendar = new GregorianCalendar(date.getZone().toTimeZone());
        gCalendar.setTime(date.toDate());
        XMLGregorianCalendar xmlGrogerianCalendar = null;
        try {
            xmlGrogerianCalendar = DatatypeFactory.newInstance().newXMLGregorianCalendar(gCalendar);

        } catch (DatatypeConfigurationException ex) {
            log.error(ex.getMessage());
            System.out.println(ex.getStackTrace());
        }
        return xmlGrogerianCalendar;
    }

    /**
     * Get the Metric Average value.
     *
     * @param metricName
     *            Metric Name parameter.
     * @return Metric average Value.
     */
    private double getMetricAverageValue(final String metricName) {
        int min = 0;
        int max = 1;

        if (metricName.equals(Constants.CPU_UTILIZATION)) {
            min = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.CPU_UTILIZATION_RANGE_AVERAGE).split(",")[0]);
            max = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.CPU_UTILIZATION_RANGE_AVERAGE).split(",")[1]);
        } else if (metricName.equals(Constants.DISK_READ_BYTES)) {
            min = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.DISK_READ_BYTES_RANGE_AVERAGE).split(",")[0]);
            max = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.DISK_READ_BYTES_RANGE_AVERAGE).split(",")[1]);
        } else if (metricName.equals(Constants.DISK_READ_OPS)) {
            min = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.DISK_READ_OPS_RANGE_AVERAGE).split(",")[0]);
            max = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.DISK_READ_OPS_RANGE_AVERAGE).split(",")[1]);
        } else if (metricName.equals(Constants.DISK_WRITE_BYTES)) {
            min = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.DISK_WRITE_BYTES_RANGE_AVERAGE).split(",")[0]);
            max = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.DISK_WRITE_BYTES_RANGE_AVERAGE).split(",")[1]);
        } else if (metricName.equals(Constants.DISK_WRITE_OPS)) {
            min = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.DISK_WRITE_OPS_RANGE_AVERAGE).split(",")[0]);
            max = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.DISK_WRITE_OPS_RANGE_AVERAGE).split(",")[1]);
        } else if (metricName.equals(Constants.NETWORK_IN)) {
            min = Integer.parseInt(
                    PropertiesUtils.getProperty(Constants.NETWORK_IN_RANGE_AVERAGE).split(",")[0]);
            max = Integer.parseInt(
                    PropertiesUtils.getProperty(Constants.NETWORK_IN_RANGE_AVERAGE).split(",")[1]);
        } else if (metricName.equals(Constants.NETWORK_OUT)) {
            min = Integer.parseInt(
                    PropertiesUtils.getProperty(Constants.NETWORK_OUT_RANGE_AVERAGE).split(",")[0]);
            max = Integer.parseInt(
                    PropertiesUtils.getProperty(Constants.NETWORK_OUT_RANGE_AVERAGE).split(",")[1]);
        } else if (metricName.equals(Constants.NETWORK_PACKETS_IN)) {
            min = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.NETWORK_PACKETS_IN_RANGE_AVERAGE).split(",")[0]);
            max = Integer
                    .parseInt(PropertiesUtils
                            .getProperty(Constants.NETWORK_PACKETS_IN_RANGE_AVERAGE).split(",")[1]);
        } else if (metricName.equals(Constants.NETWORK_PACKETS_OUT)) {
            min = Integer
                    .parseInt(
                            PropertiesUtils.getProperty(Constants.NETWORK_PACKETS_OUT_RANGE_AVERAGE)
                                    .split(",")[0]);
            max = Integer
                    .parseInt(
                            PropertiesUtils.getProperty(Constants.NETWORK_PACKETS_OUT_RANGE_AVERAGE)
                                    .split(",")[1]);
        } else if (metricName.equals(Constants.STATUS_CHECK_FAILED)) {
            min = Integer
                    .parseInt(
                            PropertiesUtils.getProperty(Constants.STATUS_CHECK_FAILED_RANGE_AVERAGE)
                                    .split(",")[0]);
            max = Integer
                    .parseInt(
                            PropertiesUtils.getProperty(Constants.STATUS_CHECK_FAILED_RANGE_AVERAGE)
                                    .split(",")[1]);
        } else if (metricName.equals(Constants.STATUS_CHECK_FAILED_INSTANCE)) {
            min = Integer.parseInt(
                    PropertiesUtils
                            .getProperty(Constants.STATUS_CHECK_FAILED_INSTANCE_RANGE_AVERAGE)
                            .split(",")[0]);
            max = Integer.parseInt(
                    PropertiesUtils
                            .getProperty(Constants.STATUS_CHECK_FAILED_INSTANCE_RANGE_AVERAGE)
                            .split(",")[1]);
        } else if (metricName.equals(Constants.STATUS_CHECK_FAILED_SYSTEM)) {
            min = Integer.parseInt(
                    PropertiesUtils.getProperty(Constants.STATUS_CHECK_FAILED_SYSTEM_RANGE_AVERAGE)
                            .split(",")[0]);
            max = Integer.parseInt(
                    PropertiesUtils.getProperty(Constants.STATUS_CHECK_FAILED_SYSTEM_RANGE_AVERAGE)
                            .split(",")[1]);
        } else if (metricName.equals(Constants.CPU_CREDIT_USAGE)) {
            min = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.CPU_CREDIT_USAGE_RANGE_AVERAGE).split(",")[0]);
            max = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.CPU_CREDIT_USAGE_RANGE_AVERAGE).split(",")[1]);
        } else if (metricName.equals(Constants.ESTIMATED_CHARGES)) {
            min = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.ESTIMATED_CHARGES_RANGE_AVERAGE).split(",")[0]);
            max = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.ESTIMATED_CHARGES_RANGE_AVERAGE).split(",")[1]);
        } else if (metricName.equals(Constants.CPU_CREDIT_BALANCE)) {
            min = Integer
                    .parseInt(PropertiesUtils
                            .getProperty(Constants.CPU_CREDIT_BALANCE_RANGE_AVERAGE).split(",")[0]);
            max = Integer
                    .parseInt(PropertiesUtils
                            .getProperty(Constants.CPU_CREDIT_BALANCE_RANGE_AVERAGE).split(",")[1]);
        }

        return new Random().nextInt((max - min) + 1) + min;
    }

    /**
     * Get the Metric Sample Count value.
     *
     * @param metricName
     *            Metric Name parameter.
     * @return Metric Sample Count Value.
     */
    private double getMetricSampleCountValue(final String metricName) {
        int min = 0;
        int max = 1;

        if (metricName.equals(Constants.CPU_UTILIZATION)) {
            min = Integer
                    .parseInt(
                            PropertiesUtils.getProperty(Constants.CPU_UTILIZATION_RANGE_SAMPLECOUNT)
                                    .split(",")[0]);
            max = Integer
                    .parseInt(
                            PropertiesUtils.getProperty(Constants.CPU_UTILIZATION_RANGE_SAMPLECOUNT)
                                    .split(",")[1]);
        } else if (metricName.equals(Constants.DISK_READ_BYTES)) {
            min = Integer
                    .parseInt(
                            PropertiesUtils.getProperty(Constants.DISK_READ_BYTES_RANGE_SAMPLECOUNT)
                                    .split(",")[0]);
            max = Integer
                    .parseInt(
                            PropertiesUtils.getProperty(Constants.DISK_READ_BYTES_RANGE_SAMPLECOUNT)
                                    .split(",")[1]);
        } else if (metricName.equals(Constants.DISK_READ_OPS)) {
            min = Integer
                    .parseInt(PropertiesUtils.getProperty(Constants.DISK_READ_OPS_RANGE_SAMPLECOUNT)
                            .split(",")[0]);
            max = Integer
                    .parseInt(PropertiesUtils.getProperty(Constants.DISK_READ_OPS_RANGE_SAMPLECOUNT)
                            .split(",")[1]);
        } else if (metricName.equals(Constants.DISK_WRITE_BYTES)) {
            min = Integer
                    .parseInt(PropertiesUtils
                            .getProperty(Constants.DISK_WRITE_BYTES_RANGE_SAMPLECOUNT)
                            .split(",")[0]);
            max = Integer
                    .parseInt(PropertiesUtils
                            .getProperty(Constants.DISK_WRITE_BYTES_RANGE_SAMPLECOUNT)
                            .split(",")[1]);
        } else if (metricName.equals(Constants.DISK_WRITE_OPS)) {
            min = Integer
                    .parseInt(PropertiesUtils
                            .getProperty(Constants.DISK_WRITE_OPS_RANGE_SAMPLECOUNT).split(",")[0]);
            max = Integer
                    .parseInt(PropertiesUtils
                            .getProperty(Constants.DISK_WRITE_OPS_RANGE_SAMPLECOUNT).split(",")[1]);
        } else if (metricName.equals(Constants.NETWORK_IN)) {
            min = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.NETWORK_IN_RANGE_SAMPLECOUNT).split(",")[0]);
            max = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.NETWORK_IN_RANGE_SAMPLECOUNT).split(",")[1]);
        } else if (metricName.equals(Constants.NETWORK_OUT)) {
            min = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.NETWORK_OUT_RANGE_SAMPLECOUNT).split(",")[0]);
            max = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.NETWORK_OUT_RANGE_SAMPLECOUNT).split(",")[1]);
        } else if (metricName.equals(Constants.NETWORK_PACKETS_IN)) {
            min = Integer.parseInt(
                    PropertiesUtils.getProperty(Constants.NETWORK_PACKETS_IN_RANGE_SAMPLECOUNT)
                            .split(",")[0]);
            max = Integer.parseInt(
                    PropertiesUtils.getProperty(Constants.NETWORK_PACKETS_IN_RANGE_SAMPLECOUNT)
                            .split(",")[1]);
        } else if (metricName.equals(Constants.NETWORK_PACKETS_OUT)) {
            min = Integer.parseInt(
                    PropertiesUtils.getProperty(Constants.NETWORK_PACKETS_OUT_RANGE_SAMPLECOUNT)
                            .split(",")[0]);
            max = Integer.parseInt(
                    PropertiesUtils.getProperty(Constants.NETWORK_PACKETS_OUT_RANGE_SAMPLECOUNT)
                            .split(",")[1]);
        } else if (metricName.equals(Constants.STATUS_CHECK_FAILED)) {
            min = Integer.parseInt(
                    PropertiesUtils.getProperty(Constants.STATUS_CHECK_FAILED_RANGE_SAMPLECOUNT)
                            .split(",")[0]);
            max = Integer.parseInt(
                    PropertiesUtils.getProperty(Constants.STATUS_CHECK_FAILED_RANGE_SAMPLECOUNT)
                            .split(",")[1]);
        } else if (metricName.equals(Constants.STATUS_CHECK_FAILED_INSTANCE)) {
            min = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.STATUS_CHECK_FAILED_INSTANCE_RANGE_SAMPLECOUNT)
                    .split(",")[0]);
            max = Integer.parseInt(PropertiesUtils
                    .getProperty(Constants.STATUS_CHECK_FAILED_INSTANCE_RANGE_SAMPLECOUNT)
                    .split(",")[1]);
        } else if (metricName.equals(Constants.STATUS_CHECK_FAILED_SYSTEM)) {
            min = Integer.parseInt(
                    PropertiesUtils
                            .getProperty(Constants.STATUS_CHECK_FAILED_SYSTEM_RANGE_SAMPLECOUNT)
                            .split(",")[0]);
            max = Integer.parseInt(
                    PropertiesUtils
                            .getProperty(Constants.STATUS_CHECK_FAILED_SYSTEM_RANGE_SAMPLECOUNT)
                            .split(",")[1]);
        } else if (metricName.equals(Constants.CPU_CREDIT_USAGE)) {
            min = Integer
                    .parseInt(PropertiesUtils
                            .getProperty(Constants.CPU_CREDIT_USAGE_RANGE_SAMPLECOUNT)
                            .split(",")[0]);
            max = Integer
                    .parseInt(PropertiesUtils
                            .getProperty(Constants.CPU_CREDIT_USAGE_RANGE_SAMPLECOUNT)
                            .split(",")[1]);
        } else if (metricName.equals(Constants.ESTIMATED_CHARGES)) {
            min = Integer.parseInt(PropertiesUtils.getProperty(
                    Constants.ESTIMATED_CHARGES_RANGE_SAMPLECOUNT).split(",")[0]);
            max = Integer.parseInt(PropertiesUtils.getProperty(
                    Constants.ESTIMATED_CHARGES_RANGE_SAMPLECOUNT).split(",")[1]);
        } else if (metricName.equals(Constants.CPU_CREDIT_BALANCE)) {
            min = Integer.parseInt(
                    PropertiesUtils.getProperty(Constants.CPU_CREDIT_BALANCE_RANGE_SAMPLECOUNT)
                            .split(",")[0]);
            max = Integer.parseInt(
                    PropertiesUtils.getProperty(Constants.CPU_CREDIT_BALANCE_RANGE_SAMPLECOUNT)
                            .split(",")[1]);
        }

        return new Random().nextInt((max - min) + 1) + min;
    }

    /**
     * Get the Metric Unit.
     *
     * @param metricName
     *            Metric Name parameter.
     * @return Standard Unit Value.
     */
    private StandardUnit getMetricUnit(final String metricName) {
        String result = null;

        if (metricName.equals(Constants.CPU_UTILIZATION)) {
            result = Constants.UNIT_PERCENT;
        } else if (metricName.equals(Constants.DISK_READ_BYTES)) {
            result = Constants.UNIT_BYTES;
        } else if (metricName.equals(Constants.DISK_READ_OPS)) {
            result = Constants.UNIT_COUNT;
        } else if (metricName.equals(Constants.DISK_WRITE_BYTES)) {
            result = Constants.UNIT_BYTES;
        } else if (metricName.equals(Constants.DISK_WRITE_OPS)) {
            result = Constants.UNIT_COUNT;
        } else if (metricName.equals(Constants.NETWORK_IN)) {
            result = Constants.UNIT_BYTES;
        } else if (metricName.equals(Constants.NETWORK_OUT)) {
            result = Constants.UNIT_BYTES;
        } else if (metricName.equals(Constants.NETWORK_PACKETS_IN)) {
            result = Constants.UNIT_COUNT;
        } else if (metricName.equals(Constants.NETWORK_PACKETS_OUT)) {
            result = Constants.UNIT_COUNT;
        } else if (metricName.equals(Constants.STATUS_CHECK_FAILED)) {
            result = Constants.UNIT_COUNT;
        } else if (metricName.equals(Constants.STATUS_CHECK_FAILED_INSTANCE)) {
            result = Constants.UNIT_COUNT;
        } else if (metricName.equals(Constants.STATUS_CHECK_FAILED_SYSTEM)) {
            result = Constants.UNIT_COUNT;
        } else if (metricName.equals(Constants.CPU_CREDIT_USAGE)) {
            result = Constants.UNIT_COUNT;
        } else if (metricName.equals(Constants.ESTIMATED_CHARGES)) {
            result = Constants.UNIT_COUNT;
        } else if (metricName.equals(Constants.CPU_CREDIT_BALANCE)) {
            result = Constants.UNIT_COUNT;
        }

        return StandardUnit.fromValue(result);
    }

    /**
     * Generate error response body in xml and write it with writer.
     *
     * @param errorCode
     *            the error code wrapped in the xml response
     * @param errorMessage
     *            the error message wrapped in the xml response
     * @return xml body for an error message which can be recognized by AWS
     *         clients
     */
    private String getXmlError(final String errorCode, final String errorMessage) {
        Map<String, Object> data = new HashMap<String, Object>();
        data.put("errorCode", StringEscapeUtils.escapeXml(errorCode));
        data.put("errorMessage", StringEscapeUtils.escapeXml(errorMessage));
        // fake a random UUID as request ID
        data.put("requestID", UUID.randomUUID().toString());

        String ret = null;

        try {
            ret = TemplateUtils.get(ERROR_RESPONSE_TEMPLATE, data);
        } catch (AwsMockException e) {
            log.error("fatal exception caught: {}", e.getMessage());
            e.printStackTrace();
        }
        return ret;
    }

    /**
     * Generate blank response body in xml and write it with writer.
     *
     * @return xml body blank response which can be recognized by AWS clients
     */
    private String getBlankResponseXml() {
        Map<String, Object> data = new HashMap<String, Object>();
        data.put("requestID", UUID.randomUUID().toString());

        String ret = null;

        try {
            ret = TemplateUtils.get(BLANK_RESPONSE_TEMPLATE, data);
        } catch (AwsMockException e) {
            log.error("fatal exception caught: {}", e.getMessage());
            e.printStackTrace();
        }
        return ret;
    }

}


text:
import os
import typing
from photomanager.db.dbutils import get_db_session, close_db_session
from photomanager.lib.pmconst import PMDBNAME, PATH_SEP
from photomanager.db.config import Config

class Command(object):
    def __init__(self, folder: str, params: typing.Dict):
        self.folder = os.path.expanduser(folder.rstrip("/\\"))
        self.params = params
        self.db_session = get_db_session(self.folder + PATH_SEP + PMDBNAME)
        self.config = Config(self.db_session)

    def do(self):
        raise NotImplementedError

    def __del__(self):
        close_db_session(self.db_session)

redpajama_set_name: RedPajamaGithub


text:
namespace webm {

Status SkipParser::Init(const ElementMetadata& metadata,
                        std::uint64_t max_size) {
  assert(metadata.size == kUnknownElementSize || metadata.size <= max_size);

  if (metadata.size == kUnknownElementSize) {
    return Status(Status::kInvalidElementSize);
  }

  num_bytes_remaining_ = metadata.size;

  return Status(Status::kOkCompleted);
}

Status SkipParser::Feed(Callback* callback, Reader* reader,
                        std::uint64_t* num_bytes_read) {
  assert(callback != nullptr);
  assert(reader != nullptr);
  assert(num_bytes_read != nullptr);

  *num_bytes_read = 0;

  if (num_bytes_remaining_ == 0) {
    return Status(Status::kOkCompleted);
  }

  Status status;
  do {
    std::uint64_t local_num_bytes_read = 0;
    status = reader->Skip(num_bytes_remaining_, &local_num_bytes_read);
    assert((status.completed_ok() &&
            local_num_bytes_read == num_bytes_remaining_) ||
           (status.ok() && local_num_bytes_read < num_bytes_remaining_) ||
           (!status.ok() && local_num_bytes_read == 0));
    *num_bytes_read += local_num_bytes_read;
    num_bytes_remaining_ -= local_num_bytes_read;
  } while (status.code == Status::kOkPartial);

  return status;
}

}  // namespace webm


text:
namespace Google.Cloud.AIPlatform.V1.Snippets
{
    // [START aiplatform_v1_generated_TensorboardService_BatchCreateTensorboardTimeSeries_sync]
    using Google.Cloud.AIPlatform.V1;

    public sealed partial class GeneratedTensorboardServiceClientSnippets
    {
        /// <summary>Snippet for BatchCreateTensorboardTimeSeries</summary>
        /// <remarks>
        /// This snippet has been automatically generated and should be regarded as a code template only.
        /// It will require modifications to work:
        /// - It may require correct/in-range values for request initialization.
        /// - It may require specifying regional endpoints when creating the service client as shown in
        ///   https://cloud.google.com/dotnet/docs/reference/help/client-configuration#endpoint.
        /// </remarks>
        public void BatchCreateTensorboardTimeSeriesRequestObject()
        {
            // Create client
            TensorboardServiceClient tensorboardServiceClient = TensorboardServiceClient.Create();
            // Initialize request argument(s)
            BatchCreateTensorboardTimeSeriesRequest request = new BatchCreateTensorboardTimeSeriesRequest
            {
                ParentAsTensorboardExperimentName = TensorboardExperimentName.FromProjectLocationTensorboardExperiment("[PROJECT]", "[LOCATION]", "[TENSORBOARD]", "[EXPERIMENT]"),
                Requests =
                {
                    new CreateTensorboardTimeSeriesRequest(),
                },
            };
            // Make the request
            BatchCreateTensorboardTimeSeriesResponse response = tensorboardServiceClient.BatchCreateTensorboardTimeSeries(request);
        }
    }
    // [END aiplatform_v1_generated_TensorboardService_BatchCreateTensorboardTimeSeries_sync]
}


text:
A (formerly?) simple shell based alternative for `locate`

These scripts are intended to be drop-in for hosts that don't have `locate` or a `locate` type package e.g. `mlocate` or `slocate`.

`locate`, `mlocate`, `slocate` and any others of their ilk use a database for historical reasons.  Back in the early 80's when `locate` was first designed, hard drive space was in short supply, and so the database format was primarily a choice made to save space.  In testing back then, the first `locate` + db combo was supposedly also faster than `grep` + plaintext.

It's 2016 now.  Hard drive space is running rampant, so we can do away with the database and use plaintext, and then that of course means we can simply use `egrep` as our frontend.  I'm very sure that other people have done this, and in `perl`, `python` or whatever else.  This is simply my attempt at it.

## slocate?  mlocate?  What?
The short version is that `slocate` tries to prevent you from getting search results for files you wouldn't have access to.  You don't want the users seeing `/root/people_to_murder.txt`, for example.  So, obviously the `s` stands for *secure*.

With `mlocate` it seeks to advance on `slocate` by *merging* differences that it finds.  In other words, it checks each directory it finds against its own db, and if the modified timestamp differs, it reindexes that directory.  At a certain scale (large filesystems, millions of files) merging gives a significant performance boost.

There was another one called `rlocate` which, if my memory serves, was a kernel module that hooked into filesystem events and essentially kept the db updated in near real-time.  Hence the `r`.

### Does simple-locate do any of that?
`simple-locate` attempts, in a rudamentary way, to replicate `slocate` and `mlocate` functionality.  It keeps the following indices:

* locate.root - an index of everything, and only root (or sufficient sudo) can search against it.
* locate.user - an index of everything that the user 'nobody' can get to.  What 'nobody' can get to, exactly, differs per system.  This is the default index for non-root users.
* locate.username (e.g. locate.rawiri) - if you run the update script as yourself, it will generate an index of files that you can access.  If you have your own index, you will search against it.  If not, you'll search against locate.user

Merging can be enabled or disabled in the update script.  As mentioned earlier, at some point of scale it will be faster.  By default, on a quiet system, it's maybe 30% slower.  This is down to a few engineering choices forcing the use of loops.

## Installation

To install, simply drop these scripts into an appropriate place in your `$PATH` e.g. `/usr/local/bin`, then obviously ensure they are executable with `chmod +x`.  Next, setup a `cron` entry to your liking for the update script.  Optionally go through the update script and double check that the variables are to your liking.  Then you could optionally manually run it.

If you don't have an existing `locate` binary, you can rename `simple-locate` to `locate`, or symlink it.  These scripts will otherwise not conflict with, and will happily live side by side with a classic flavour of `locate`.

## Configuration
There's not much to configure, but if you like you can edit the update script.  All the comments you need are there.

## Usage
`simple-locate` is basically a veneer thin wrapper for `egrep`, so `egrep` syntax, rather than `locate` applies.  All options available to `egrep` apply to `simple-locate` too, so for example you can do a case-insensitive search using `-i`.

## Compatibility
So far this has been tested on CentOS 4.7, Centos 6.7, Ubuntu 16.04, FreeBSD 7, Solaris 9 and Solaris 11.  Updates have been applied to suit while maintaining relative portability.  I'll be interested to hear if it works elsewhere, specifically on AIX and/or HPUX.  I'm also interested in any feedback on issues etc.

For the record: FreeBSD provided some interesting challenges and solutions (e.g. `chown 0:0`), but Solaris 9 has forced the most regressions.

### A note about 'find' compatibility
In order to try and eke out some performance, `simple-locate` directs `find` to ignore certain filesystem types and paths.  A full command might look something like this:

```
find / \( -fstype afs -o -fstype autofs -o -fstype coda -o -fstype ctfs -o -fstype dev -o -fstype devfs -o -fstype devpts -o -fstype fd -o -fstype ftpfs -o -fstype hsfs -o -fstype iso9660 -o -fstype lofs -o -fstype mfs -o -fstype mntfs -o -fstype namefs -o -fstype ncpfs -o -fstype nfs -o -fstype NFS -o -fstype objfs -o -fstype pcfs -o -fstype proc -o -fstype shfs -o -fstype smbfs -o -fstype sysfs -o -fstype tmpfs -o -fstype uvfs -o -path /cdrom -o -path /media -o -path /proc -o -path /tmp -o -path /var/tmp \) -prune -o -type d -mtime -1 -print
```

Unfortunately, not all versions of `find` support `-path`, primarily Solaris and AIX.  I have tested various ways that claim to be portable workarounds and here's what I found:

`-o -type d -name '/path'`

This one upsets GNU `find` which prints an error.  Slashes might not be interpreted, which means you can't exclude e.g. `/var/tmp`, so instead you have to exclude all directories named `tmp`, which might not be what you want.  You lose flexibility and risk having unintended matches, which are then excluded.

`-o -inum $inum_of_/path`

This one had a lot of promise.  It is fast and portable but it's unpredictable.  It works on the premise of excluding by inode number, which can be easily found e.g. `ls -di /some/path | awk '{print $1}'`.  Unfortunately, in testing on a Solaris 9 host, it showed that multiple filesystems shared an inode, and so we had unintended matches again.

```
sol9test{root}: ls -di /proc
         2 /proc
sol9test{root}: ls -di /opt
         2 /opt
sol9test{root}: ls -di /home
         2 /home
```

See?  We want to exclude `/proc`, it has an inode number of `2`, but so does `/opt` and `/home` at least, which we do want to index.  So let's move on...

`-o -exec test "{}" = "/path" \;`

Last up is this one.  It is brutally slow compared to all other options, but it's reliable.  Do NOT chain it like this:

`-o -exec test "{}" = "/path" -o -exec test "{}" = "/some/other/path" \;`

Instead, chain it like this:

`-o -exec test "{}" = "/path" -o "{}" = "/some/other/path" \;`

To give you an example, in a test using the full chaining method, on my SSD, finding directories modified in the last 24 hours, we get:

```
real	12m11.409s
user	0m16.300s
sys	1m52.700s
```

Using the more compact chaining method, however:

```
real	1m45.280s
user	0m5.744s
sys	0m19.556s
```

These results are repeatable, no matter which order etc you run the commands in.  They will obviously differ per system and day.

So ultimately, `simple-locate` will test whether `find` can use `-path`, and if so, use it.  It will otherwise failover to the slower `-exec test` method.  If, on Solaris or AIX, you can provide GNU `find`, then that will be to your benefit.  Not just for `simple-locate`.

## Performance
### Searching
On my workstation, which has an SSD, let's test `find`:

```
# time find / -type d -name "bin"
...
real	0m16.184s
user	0m2.224s
sys	0m5.888s
```

Followed by `locate` (i.e. `mlocate`):
```
# time locate -r '.*/bin$'
...
real	0m9.778s
user	0m9.656s
sys	0m0.032s
```

Followed by `simple-locate`:
```
# time ./smlocate /bin$
...
real	0m0.074s
user	0m0.028s
sys	0m0.028s
```
### Indexing
`updatedb` (`mlocate`):

```
# time updatedb

real	0m6.218s
user	0m0.428s
sys	0m2.184s
```

`simple-locate` (merging disabled)
```
# time ./update-smlocate

real	0m9.422s
user	0m6.992s
sys	0m2.056s
```

I'd be lying if I said this was a reliable test result.  Realistically it varies depending on system load.  Here's the worst test result:

```
# time ./update-smlocate

real	0m37.312s
user	0m10.288s
sys	0m11.120s
```

Which, for an overnight `cron` job isn't the end of the world anyway.

`simple-locate` (merging enabled)
```
# time ./update-smlocate

real	0m22.068s
user	0m10.672s
sys	0m4.528s
```

As we can tell, at this scale, merging is slightly punitive over our non-merging best-case, but in repeated testing it seems to stick around the 19-22s mark.

## Known Problems
Most problems have been dealt with, the major remaining one that I'm aware of is duplication in merging.  For example, let's say that the update script finds the following directories as having been modified in the last 24 hours:

* `/home/userA`
* `/home/userA/.local/someApp/cache`
* `/home/userA/.secret/stashedfiles/`

So the update script will filter `^/home/userA` out of the index files, which covers the other two directories.  It will then reindex `/home/userA`, which includes the other two directories.  But in due course it will reindex the other two directories as well.

I currently have no great incentive to fix this, for two reasons:

* Indexing a directory is pretty fast.  While it's inelegant, there's no great performance loss with the reindexing.  The main performance hit is in generating the list of directories modified in the last 24 hours
* The searching script runs `sort` and `uniq` across its output anyway, which removes duplicates.

This is a "maybe on a rainy day" fix.


text:
package org.apache.spark

import scala.collection.mutable.ArrayBuffer

import org.mockito.ArgumentMatchers.any
import org.mockito.Mockito._
import org.roaringbitmap.RoaringBitmap

import org.apache.spark.LocalSparkContext._
import org.apache.spark.broadcast.BroadcastManager
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.Network.{RPC_ASK_TIMEOUT, RPC_MESSAGE_MAX_SIZE}
import org.apache.spark.internal.config.Tests.IS_TESTING
import org.apache.spark.rpc.{RpcAddress, RpcCallContext, RpcEnv}
import org.apache.spark.scheduler.{CompressedMapStatus, MapStatus, MergeStatus}
import org.apache.spark.shuffle.FetchFailedException
import org.apache.spark.storage.{BlockManagerId, ShuffleBlockId, ShuffleMergedBlockId}

class MapOutputTrackerSuite extends SparkFunSuite with LocalSparkContext {
  private val conf = new SparkConf

  private def newTrackerMaster(sparkConf: SparkConf = conf) = {
    val broadcastManager = new BroadcastManager(true, sparkConf)
    new MapOutputTrackerMaster(sparkConf, broadcastManager, true)
  }

  def createRpcEnv(name: String, host: String = "localhost", port: Int = 0,
      securityManager: SecurityManager = new SecurityManager(conf)): RpcEnv = {
    RpcEnv.create(name, host, port, conf, securityManager)
  }

  test("master start and stop") {
    val rpcEnv = createRpcEnv("test")
    val tracker = newTrackerMaster()
    tracker.trackerEndpoint = rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME,
      new MapOutputTrackerMasterEndpoint(rpcEnv, tracker, conf))
    tracker.stop()
    rpcEnv.shutdown()
  }

  test("master register shuffle and fetch") {
    val rpcEnv = createRpcEnv("test")
    val tracker = newTrackerMaster()
    tracker.trackerEndpoint = rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME,
      new MapOutputTrackerMasterEndpoint(rpcEnv, tracker, conf))
    tracker.registerShuffle(10, 2, MergeStatus.SHUFFLE_PUSH_DUMMY_NUM_REDUCES)
    assert(tracker.containsShuffle(10))
    val size1000 = MapStatus.decompressSize(MapStatus.compressSize(1000L))
    val size10000 = MapStatus.decompressSize(MapStatus.compressSize(10000L))
    tracker.registerMapOutput(10, 0, MapStatus(BlockManagerId("a", "hostA", 1000),
        Array(1000L, 10000L), 5))
    tracker.registerMapOutput(10, 1, MapStatus(BlockManagerId("b", "hostB", 1000),
        Array(10000L, 1000L), 6))
    val statuses = tracker.getMapSizesByExecutorId(10, 0)
    assert(statuses.toSet ===
      Seq((BlockManagerId("a", "hostA", 1000),
        ArrayBuffer((ShuffleBlockId(10, 5, 0), size1000, 0))),
          (BlockManagerId("b", "hostB", 1000),
            ArrayBuffer((ShuffleBlockId(10, 6, 0), size10000, 1)))).toSet)
    assert(0 == tracker.getNumCachedSerializedBroadcast)
    tracker.stop()
    rpcEnv.shutdown()
  }

  test("master register and unregister shuffle") {
    val rpcEnv = createRpcEnv("test")
    val tracker = newTrackerMaster()
    tracker.trackerEndpoint = rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME,
      new MapOutputTrackerMasterEndpoint(rpcEnv, tracker, conf))
    tracker.registerShuffle(10, 2, MergeStatus.SHUFFLE_PUSH_DUMMY_NUM_REDUCES)
    val compressedSize1000 = MapStatus.compressSize(1000L)
    val compressedSize10000 = MapStatus.compressSize(10000L)
    tracker.registerMapOutput(10, 0, MapStatus(BlockManagerId("a", "hostA", 1000),
      Array(compressedSize1000, compressedSize10000), 5))
    tracker.registerMapOutput(10, 1, MapStatus(BlockManagerId("b", "hostB", 1000),
      Array(compressedSize10000, compressedSize1000), 6))
    assert(tracker.containsShuffle(10))
    assert(tracker.getMapSizesByExecutorId(10, 0).nonEmpty)
    assert(0 == tracker.getNumCachedSerializedBroadcast)
    tracker.unregisterShuffle(10)
    assert(!tracker.containsShuffle(10))
    assert(tracker.getMapSizesByExecutorId(10, 0).isEmpty)

    tracker.stop()
    rpcEnv.shutdown()
  }

  test("master register shuffle and unregister map output and fetch") {
    val rpcEnv = createRpcEnv("test")
    val tracker = newTrackerMaster()
    tracker.trackerEndpoint = rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME,
      new MapOutputTrackerMasterEndpoint(rpcEnv, tracker, conf))
    tracker.registerShuffle(10, 2, MergeStatus.SHUFFLE_PUSH_DUMMY_NUM_REDUCES)
    val compressedSize1000 = MapStatus.compressSize(1000L)
    val compressedSize10000 = MapStatus.compressSize(10000L)
    tracker.registerMapOutput(10, 0, MapStatus(BlockManagerId("a", "hostA", 1000),
        Array(compressedSize1000, compressedSize1000, compressedSize1000), 5))
    tracker.registerMapOutput(10, 1, MapStatus(BlockManagerId("b", "hostB", 1000),
        Array(compressedSize10000, compressedSize1000, compressedSize1000), 6))

    assert(0 == tracker.getNumCachedSerializedBroadcast)
    // As if we had two simultaneous fetch failures
    tracker.unregisterMapOutput(10, 0, BlockManagerId("a", "hostA", 1000))
    tracker.unregisterMapOutput(10, 0, BlockManagerId("a", "hostA", 1000))

    // The remaining reduce task might try to grab the output despite the shuffle failure;
    // this should cause it to fail, and the scheduler will ignore the failure due to the
    // stage already being aborted.
    intercept[FetchFailedException] { tracker.getMapSizesByExecutorId(10, 1) }

    tracker.stop()
    rpcEnv.shutdown()
  }

  test("remote fetch") {
    val hostname = "localhost"
    val rpcEnv = createRpcEnv("spark", hostname, 0, new SecurityManager(conf))

    val masterTracker = newTrackerMaster()
    masterTracker.trackerEndpoint = rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME,
      new MapOutputTrackerMasterEndpoint(rpcEnv, masterTracker, conf))

    val mapWorkerRpcEnv = createRpcEnv("spark-worker", hostname, 0, new SecurityManager(conf))
    val mapWorkerTracker = new MapOutputTrackerWorker(conf)
    mapWorkerTracker.trackerEndpoint =
      mapWorkerRpcEnv.setupEndpointRef(rpcEnv.address, MapOutputTracker.ENDPOINT_NAME)

    masterTracker.registerShuffle(10, 1, MergeStatus.SHUFFLE_PUSH_DUMMY_NUM_REDUCES)
    mapWorkerTracker.updateEpoch(masterTracker.getEpoch)
    // This is expected to fail because no outputs have been registered for the shuffle.
    intercept[FetchFailedException] { mapWorkerTracker.getMapSizesByExecutorId(10, 0) }

    val size1000 = MapStatus.decompressSize(MapStatus.compressSize(1000L))
    masterTracker.registerMapOutput(10, 0, MapStatus(
      BlockManagerId("a", "hostA", 1000), Array(1000L), 5))
    mapWorkerTracker.updateEpoch(masterTracker.getEpoch)
    assert(mapWorkerTracker.getMapSizesByExecutorId(10, 0).toSeq ===
      Seq((BlockManagerId("a", "hostA", 1000),
        ArrayBuffer((ShuffleBlockId(10, 5, 0), size1000, 0)))))
    assert(0 == masterTracker.getNumCachedSerializedBroadcast)

    val masterTrackerEpochBeforeLossOfMapOutput = masterTracker.getEpoch
    masterTracker.unregisterMapOutput(10, 0, BlockManagerId("a", "hostA", 1000))
    assert(masterTracker.getEpoch > masterTrackerEpochBeforeLossOfMapOutput)
    mapWorkerTracker.updateEpoch(masterTracker.getEpoch)
    intercept[FetchFailedException] { mapWorkerTracker.getMapSizesByExecutorId(10, 0) }

    // failure should be cached
    intercept[FetchFailedException] { mapWorkerTracker.getMapSizesByExecutorId(10, 0) }
    assert(0 == masterTracker.getNumCachedSerializedBroadcast)

    masterTracker.stop()
    mapWorkerTracker.stop()
    rpcEnv.shutdown()
    mapWorkerRpcEnv.shutdown()
  }

  test("remote fetch below max RPC message size") {
    val newConf = new SparkConf
    newConf.set(RPC_MESSAGE_MAX_SIZE, 1)
    newConf.set(RPC_ASK_TIMEOUT, "1") // Fail fast
    newConf.set(SHUFFLE_MAPOUTPUT_MIN_SIZE_FOR_BROADCAST, 1048576L)

    val masterTracker = newTrackerMaster(newConf)
    val rpcEnv = createRpcEnv("spark")
    val masterEndpoint = new MapOutputTrackerMasterEndpoint(rpcEnv, masterTracker, newConf)
    masterTracker.trackerEndpoint =
      rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME, masterEndpoint)

    // Message size should be ~123B, and no exception should be thrown
    masterTracker.registerShuffle(10, 1, MergeStatus.SHUFFLE_PUSH_DUMMY_NUM_REDUCES)
    masterTracker.registerMapOutput(10, 0, MapStatus(
      BlockManagerId("88", "mph", 1000), Array.fill[Long](10)(0), 5))
    val senderAddress = RpcAddress("localhost", 12345)
    val rpcCallContext = mock(classOf[RpcCallContext])
    when(rpcCallContext.senderAddress).thenReturn(senderAddress)
    masterEndpoint.receiveAndReply(rpcCallContext)(GetMapOutputStatuses(10))
    // Default size for broadcast in this testsuite is set to -1 so should not cause broadcast
    // to be used.
    verify(rpcCallContext, timeout(30000)).reply(any())
    assert(0 == masterTracker.getNumCachedSerializedBroadcast)

    masterTracker.stop()
    rpcEnv.shutdown()
  }

  test("min broadcast size exceeds max RPC message size") {
    val newConf = new SparkConf
    newConf.set(RPC_MESSAGE_MAX_SIZE, 1)
    newConf.set(RPC_ASK_TIMEOUT, "1") // Fail fast
    newConf.set(SHUFFLE_MAPOUTPUT_MIN_SIZE_FOR_BROADCAST, Int.MaxValue.toLong)

    intercept[IllegalArgumentException] { newTrackerMaster(newConf) }
  }

  test("getLocationsWithLargestOutputs with multiple outputs in same machine") {
    val rpcEnv = createRpcEnv("test")
    val tracker = newTrackerMaster()
    tracker.trackerEndpoint = rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME,
      new MapOutputTrackerMasterEndpoint(rpcEnv, tracker, conf))
    // Setup 3 map tasks
    // on hostA with output size 2
    // on hostA with output size 2
    // on hostB with output size 3
    tracker.registerShuffle(10, 3, MergeStatus.SHUFFLE_PUSH_DUMMY_NUM_REDUCES)
    tracker.registerMapOutput(10, 0, MapStatus(BlockManagerId("a", "hostA", 1000),
        Array(2L), 5))
    tracker.registerMapOutput(10, 1, MapStatus(BlockManagerId("a", "hostA", 1000),
        Array(2L), 6))
    tracker.registerMapOutput(10, 2, MapStatus(BlockManagerId("b", "hostB", 1000),
        Array(3L), 7))

    // When the threshold is 50%, only host A should be returned as a preferred location
    // as it has 4 out of 7 bytes of output.
    val topLocs50 = tracker.getLocationsWithLargestOutputs(10, 0, 1, 0.5)
    assert(topLocs50.nonEmpty)
    assert(topLocs50.get.size === 1)
    assert(topLocs50.get.head === BlockManagerId("a", "hostA", 1000))

    // When the threshold is 20%, both hosts should be returned as preferred locations.
    val topLocs20 = tracker.getLocationsWithLargestOutputs(10, 0, 1, 0.2)
    assert(topLocs20.nonEmpty)
    assert(topLocs20.get.size === 2)
    assert(topLocs20.get.toSet ===
           Seq(BlockManagerId("a", "hostA", 1000), BlockManagerId("b", "hostB", 1000)).toSet)

    tracker.stop()
    rpcEnv.shutdown()
  }

  test("remote fetch using broadcast") {
    val newConf = new SparkConf
    newConf.set(RPC_MESSAGE_MAX_SIZE, 1)
    newConf.set(RPC_ASK_TIMEOUT, "1") // Fail fast
    newConf.set(SHUFFLE_MAPOUTPUT_MIN_SIZE_FOR_BROADCAST, 10240L) // 10 KiB << 1MiB framesize

    // needs TorrentBroadcast so need a SparkContext
    withSpark(new SparkContext("local", "MapOutputTrackerSuite", newConf)) { sc =>
      val masterTracker = sc.env.mapOutputTracker.asInstanceOf[MapOutputTrackerMaster]
      val rpcEnv = sc.env.rpcEnv
      val masterEndpoint = new MapOutputTrackerMasterEndpoint(rpcEnv, masterTracker, newConf)
      rpcEnv.stop(masterTracker.trackerEndpoint)
      rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME, masterEndpoint)

      // Frame size should be ~1.1MB, and MapOutputTrackerMasterEndpoint should throw exception.
      // Note that the size is hand-selected here because map output statuses are compressed before
      // being sent.
      masterTracker.registerShuffle(20, 100, MergeStatus.SHUFFLE_PUSH_DUMMY_NUM_REDUCES)
      (0 until 100).foreach { i =>
        masterTracker.registerMapOutput(20, i, new CompressedMapStatus(
          BlockManagerId("999", "mps", 1000), Array.fill[Long](4000000)(0), 5))
      }
      val senderAddress = RpcAddress("localhost", 12345)
      val rpcCallContext = mock(classOf[RpcCallContext])
      when(rpcCallContext.senderAddress).thenReturn(senderAddress)
      masterEndpoint.receiveAndReply(rpcCallContext)(GetMapOutputStatuses(20))
      // should succeed since majority of data is broadcast and actual serialized
      // message size is small
      verify(rpcCallContext, timeout(30000)).reply(any())
      assert(1 == masterTracker.getNumCachedSerializedBroadcast)
      masterTracker.unregisterShuffle(20)
      assert(0 == masterTracker.getNumCachedSerializedBroadcast)
    }
  }

  test("equally divide map statistics tasks") {
    val func = newTrackerMaster().equallyDivide _
    val cases = Seq((0, 5), (4, 5), (15, 5), (16, 5), (17, 5), (18, 5), (19, 5), (20, 5))
    val expects = Seq(
      Seq(0, 0, 0, 0, 0),
      Seq(1, 1, 1, 1, 0),
      Seq(3, 3, 3, 3, 3),
      Seq(4, 3, 3, 3, 3),
      Seq(4, 4, 3, 3, 3),
      Seq(4, 4, 4, 3, 3),
      Seq(4, 4, 4, 4, 3),
      Seq(4, 4, 4, 4, 4))
    cases.zip(expects).foreach { case ((num, divisor), expect) =>
      val answer = func(num, divisor).toSeq
      var wholeSplit = (0 until num)
      answer.zip(expect).foreach { case (split, expectSplitLength) =>
        val (currentSplit, rest) = wholeSplit.splitAt(expectSplitLength)
        assert(currentSplit.toSet == split.toSet)
        wholeSplit = rest
      }
    }
  }

  test("zero-sized blocks should be excluded when getMapSizesByExecutorId") {
    val rpcEnv = createRpcEnv("test")
    val tracker = newTrackerMaster()
    tracker.trackerEndpoint = rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME,
      new MapOutputTrackerMasterEndpoint(rpcEnv, tracker, conf))
    tracker.registerShuffle(10, 2, MergeStatus.SHUFFLE_PUSH_DUMMY_NUM_REDUCES)

    val size0 = MapStatus.decompressSize(MapStatus.compressSize(0L))
    val size1000 = MapStatus.decompressSize(MapStatus.compressSize(1000L))
    val size10000 = MapStatus.decompressSize(MapStatus.compressSize(10000L))
    tracker.registerMapOutput(10, 0, MapStatus(BlockManagerId("a", "hostA", 1000),
      Array(size0, size1000, size0, size10000), 5))
    tracker.registerMapOutput(10, 1, MapStatus(BlockManagerId("b", "hostB", 1000),
      Array(size10000, size0, size1000, size0), 6))
    assert(tracker.containsShuffle(10))
    assert(tracker.getMapSizesByExecutorId(10, 0, 2, 0, 4).toSeq ===
        Seq(
          (BlockManagerId("a", "hostA", 1000),
              Seq((ShuffleBlockId(10, 5, 1), size1000, 0),
                (ShuffleBlockId(10, 5, 3), size10000, 0))),
          (BlockManagerId("b", "hostB", 1000),
              Seq((ShuffleBlockId(10, 6, 0), size10000, 1),
                (ShuffleBlockId(10, 6, 2), size1000, 1)))
        )
    )

    tracker.unregisterShuffle(10)
    tracker.stop()
    rpcEnv.shutdown()
  }

  test("SPARK-32921: master register and unregister merge result") {
    conf.set(PUSH_BASED_SHUFFLE_ENABLED, true)
    conf.set(IS_TESTING, true)
    val rpcEnv = createRpcEnv("test")
    val tracker = newTrackerMaster()
    tracker.trackerEndpoint = rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME,
      new MapOutputTrackerMasterEndpoint(rpcEnv, tracker, conf))
    tracker.registerShuffle(10, 4, 2)
    assert(tracker.containsShuffle(10))
    val bitmap = new RoaringBitmap()
    bitmap.add(0)
    bitmap.add(1)

    tracker.registerMergeResult(10, 0, MergeStatus(BlockManagerId("a", "hostA", 1000), 0,
      bitmap, 1000L))
    tracker.registerMergeResult(10, 1, MergeStatus(BlockManagerId("b", "hostB", 1000), 0,
      bitmap, 1000L))
    assert(tracker.getNumAvailableMergeResults(10) == 2)
    tracker.unregisterMergeResult(10, 0, BlockManagerId("a", "hostA", 1000))
    assert(tracker.getNumAvailableMergeResults(10) == 1)
    tracker.stop()
    rpcEnv.shutdown()
  }

  test("SPARK-32921: get map sizes with merged shuffle") {
    conf.set(PUSH_BASED_SHUFFLE_ENABLED, true)
    conf.set(IS_TESTING, true)
    val hostname = "localhost"
    val rpcEnv = createRpcEnv("spark", hostname, 0, new SecurityManager(conf))

    val masterTracker = newTrackerMaster()
    masterTracker.trackerEndpoint = rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME,
      new MapOutputTrackerMasterEndpoint(rpcEnv, masterTracker, conf))

    val slaveRpcEnv = createRpcEnv("spark-slave", hostname, 0, new SecurityManager(conf))
    val slaveTracker = new MapOutputTrackerWorker(conf)
    slaveTracker.trackerEndpoint =
      slaveRpcEnv.setupEndpointRef(rpcEnv.address, MapOutputTracker.ENDPOINT_NAME)

    masterTracker.registerShuffle(10, 4, 1)
    slaveTracker.updateEpoch(masterTracker.getEpoch)
    val bitmap = new RoaringBitmap()
    bitmap.add(0)
    bitmap.add(1)
    bitmap.add(3)

    val blockMgrId = BlockManagerId("a", "hostA", 1000)
    masterTracker.registerMapOutput(10, 0, MapStatus(blockMgrId, Array(1000L), 0))
    masterTracker.registerMapOutput(10, 1, MapStatus(blockMgrId, Array(1000L), 1))
    masterTracker.registerMapOutput(10, 2, MapStatus(blockMgrId, Array(1000L), 2))
    masterTracker.registerMapOutput(10, 3, MapStatus(blockMgrId, Array(1000L), 3))

    masterTracker.registerMergeResult(10, 0, MergeStatus(blockMgrId, 0,
      bitmap, 3000L))
    slaveTracker.updateEpoch(masterTracker.getEpoch)
    val size1000 = MapStatus.decompressSize(MapStatus.compressSize(1000L))
    assert(slaveTracker.getMapSizesByExecutorId(10, 0).toSeq ===
      Seq((blockMgrId, ArrayBuffer((ShuffleMergedBlockId(10, 0, 0), 3000, -1),
        (ShuffleBlockId(10, 2, 0), size1000, 2)))))

    masterTracker.stop()
    slaveTracker.stop()
    rpcEnv.shutdown()
    slaveRpcEnv.shutdown()
  }

  test("SPARK-32921: get map statuses from merged shuffle") {
    conf.set(PUSH_BASED_SHUFFLE_ENABLED, true)
    conf.set(IS_TESTING, true)
    val hostname = "localhost"
    val rpcEnv = createRpcEnv("spark", hostname, 0, new SecurityManager(conf))

    val masterTracker = newTrackerMaster()
    masterTracker.trackerEndpoint = rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME,
      new MapOutputTrackerMasterEndpoint(rpcEnv, masterTracker, conf))

    val slaveRpcEnv = createRpcEnv("spark-slave", hostname, 0, new SecurityManager(conf))
    val slaveTracker = new MapOutputTrackerWorker(conf)
    slaveTracker.trackerEndpoint =
      slaveRpcEnv.setupEndpointRef(rpcEnv.address, MapOutputTracker.ENDPOINT_NAME)

    masterTracker.registerShuffle(10, 4, 1)
    slaveTracker.updateEpoch(masterTracker.getEpoch)
    // This is expected to fail because no outputs have been registered for the shuffle.
    intercept[FetchFailedException] { slaveTracker.getMapSizesByExecutorId(10, 0) }
    val bitmap = new RoaringBitmap()
    bitmap.add(0)
    bitmap.add(1)
    bitmap.add(2)
    bitmap.add(3)

    val blockMgrId = BlockManagerId("a", "hostA", 1000)
    masterTracker.registerMapOutput(10, 0, MapStatus(blockMgrId, Array(1000L), 0))
    masterTracker.registerMapOutput(10, 1, MapStatus(blockMgrId, Array(1000L), 1))
    masterTracker.registerMapOutput(10, 2, MapStatus(blockMgrId, Array(1000L), 2))
    masterTracker.registerMapOutput(10, 3, MapStatus(blockMgrId, Array(1000L), 3))

    masterTracker.registerMergeResult(10, 0, MergeStatus(blockMgrId, 0,
      bitmap, 4000L))
    slaveTracker.updateEpoch(masterTracker.getEpoch)
    val size1000 = MapStatus.decompressSize(MapStatus.compressSize(1000L))
    assert(slaveTracker.getMapSizesForMergeResult(10, 0).toSeq ===
      Seq((blockMgrId, ArrayBuffer((ShuffleBlockId(10, 0, 0), size1000, 0),
        (ShuffleBlockId(10, 1, 0), size1000, 1), (ShuffleBlockId(10, 2, 0), size1000, 2),
        (ShuffleBlockId(10, 3, 0), size1000, 3)))))
    masterTracker.stop()
    slaveTracker.stop()
    rpcEnv.shutdown()
    slaveRpcEnv.shutdown()
  }

  test("SPARK-32921: get map statuses for merged shuffle block chunks") {
    conf.set(PUSH_BASED_SHUFFLE_ENABLED, true)
    conf.set(IS_TESTING, true)
    val hostname = "localhost"
    val rpcEnv = createRpcEnv("spark", hostname, 0, new SecurityManager(conf))

    val masterTracker = newTrackerMaster()
    masterTracker.trackerEndpoint = rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME,
      new MapOutputTrackerMasterEndpoint(rpcEnv, masterTracker, conf))

    val slaveRpcEnv = createRpcEnv("spark-slave", hostname, 0, new SecurityManager(conf))
    val slaveTracker = new MapOutputTrackerWorker(conf)
    slaveTracker.trackerEndpoint =
      slaveRpcEnv.setupEndpointRef(rpcEnv.address, MapOutputTracker.ENDPOINT_NAME)

    masterTracker.registerShuffle(10, 4, 1)
    slaveTracker.updateEpoch(masterTracker.getEpoch)

    val blockMgrId = BlockManagerId("a", "hostA", 1000)
    masterTracker.registerMapOutput(10, 0, MapStatus(blockMgrId, Array(1000L), 0))
    masterTracker.registerMapOutput(10, 1, MapStatus(blockMgrId, Array(1000L), 1))
    masterTracker.registerMapOutput(10, 2, MapStatus(blockMgrId, Array(1000L), 2))
    masterTracker.registerMapOutput(10, 3, MapStatus(blockMgrId, Array(1000L), 3))

    val chunkBitmap = new RoaringBitmap()
    chunkBitmap.add(0)
    chunkBitmap.add(2)
    val size1000 = MapStatus.decompressSize(MapStatus.compressSize(1000L))
    assert(slaveTracker.getMapSizesForMergeResult(10, 0, chunkBitmap).toSeq ===
      Seq((blockMgrId, ArrayBuffer((ShuffleBlockId(10, 0, 0), size1000, 0),
        (ShuffleBlockId(10, 2, 0), size1000, 2))))
    )
    masterTracker.stop()
    slaveTracker.stop()
    rpcEnv.shutdown()
    slaveRpcEnv.shutdown()
  }

  test("SPARK-32921: getPreferredLocationsForShuffle with MergeStatus") {
    val rpcEnv = createRpcEnv("test")
    val tracker = newTrackerMaster()
    sc = new SparkContext("local", "test", conf.clone())
    tracker.trackerEndpoint = rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME,
      new MapOutputTrackerMasterEndpoint(rpcEnv, tracker, conf))
    // Setup 5 map tasks
    // on hostA with output size 2
    // on hostA with output size 2
    // on hostB with output size 3
    // on hostB with output size 3
    // on hostC with output size 1
    // on hostC with output size 1
    tracker.registerShuffle(10, 6, 1)
    tracker.registerMapOutput(10, 0, MapStatus(BlockManagerId("a", "hostA", 1000),
      Array(2L), 5))
    tracker.registerMapOutput(10, 1, MapStatus(BlockManagerId("a", "hostA", 1000),
      Array(2L), 6))
    tracker.registerMapOutput(10, 2, MapStatus(BlockManagerId("b", "hostB", 1000),
      Array(3L), 7))
    tracker.registerMapOutput(10, 3, MapStatus(BlockManagerId("b", "hostB", 1000),
      Array(3L), 8))
    tracker.registerMapOutput(10, 4, MapStatus(BlockManagerId("c", "hostC", 1000),
      Array(1L), 9))
    tracker.registerMapOutput(10, 5, MapStatus(BlockManagerId("c", "hostC", 1000),
      Array(1L), 10))

    val rdd = sc.parallelize(1 to 6, 6).map(num => (num, num).asInstanceOf[Product2[Int, Int]])
    val mockShuffleDep = mock(classOf[ShuffleDependency[Int, Int, _]])
    when(mockShuffleDep.shuffleId).thenReturn(10)
    when(mockShuffleDep.partitioner).thenReturn(new HashPartitioner(1))
    when(mockShuffleDep.rdd).thenReturn(rdd)

    // Prepare a MergeStatus that merges 4 out of 5 blocks
    val bitmap80 = new RoaringBitmap()
    bitmap80.add(0)
    bitmap80.add(1)
    bitmap80.add(2)
    bitmap80.add(3)
    bitmap80.add(4)
    tracker.registerMergeResult(10, 0, MergeStatus(BlockManagerId("a", "hostA", 1000), 0,
      bitmap80, 11))

    val preferredLocs1 = tracker.getPreferredLocationsForShuffle(mockShuffleDep, 0)
    assert(preferredLocs1.nonEmpty)
    assert(preferredLocs1.length === 1)
    assert(preferredLocs1.head === "hostA")

    tracker.unregisterMergeResult(10, 0, BlockManagerId("a", "hostA", 1000))
    // Prepare another MergeStatus that merges only 1 out of 5 blocks
    val bitmap20 = new RoaringBitmap()
    bitmap20.add(0)
    tracker.registerMergeResult(10, 0, MergeStatus(BlockManagerId("a", "hostA", 1000), 0,
      bitmap20, 2))

    val preferredLocs2 = tracker.getPreferredLocationsForShuffle(mockShuffleDep, 0)
    assert(preferredLocs2.nonEmpty)
    assert(preferredLocs2.length === 2)
    assert(preferredLocs2 === Seq("hostA", "hostB"))

    tracker.stop()
    rpcEnv.shutdown()
  }

  test("SPARK-34939: remote fetch using broadcast if broadcasted value is destroyed") {
    val newConf = new SparkConf
    newConf.set(RPC_MESSAGE_MAX_SIZE, 1)
    newConf.set(RPC_ASK_TIMEOUT, "1") // Fail fast
    newConf.set(SHUFFLE_MAPOUTPUT_MIN_SIZE_FOR_BROADCAST, 10240L) // 10 KiB << 1MiB framesize

    // needs TorrentBroadcast so need a SparkContext
    withSpark(new SparkContext("local", "MapOutputTrackerSuite", newConf)) { sc =>
      val masterTracker = sc.env.mapOutputTracker.asInstanceOf[MapOutputTrackerMaster]
      val rpcEnv = sc.env.rpcEnv
      val masterEndpoint = new MapOutputTrackerMasterEndpoint(rpcEnv, masterTracker, newConf)
      rpcEnv.stop(masterTracker.trackerEndpoint)
      rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME, masterEndpoint)

      masterTracker.registerShuffle(20, 100, MergeStatus.SHUFFLE_PUSH_DUMMY_NUM_REDUCES)
      (0 until 100).foreach { i =>
        masterTracker.registerMapOutput(20, i, new CompressedMapStatus(
          BlockManagerId("999", "mps", 1000), Array.fill[Long](4000000)(0), 5))
      }

      val mapWorkerRpcEnv = createRpcEnv("spark-worker", "localhost", 0, new SecurityManager(conf))
      val mapWorkerTracker = new MapOutputTrackerWorker(conf)
      mapWorkerTracker.trackerEndpoint =
        mapWorkerRpcEnv.setupEndpointRef(rpcEnv.address, MapOutputTracker.ENDPOINT_NAME)

      val fetchedBytes = mapWorkerTracker.trackerEndpoint
        .askSync[Array[Byte]](GetMapOutputStatuses(20))
      assert(fetchedBytes(0) == 1)

      // Normally `unregisterMapOutput` triggers the destroy of broadcasted value.
      // But the timing of destroying broadcasted value is indeterminate, we manually destroy
      // it by blocking.
      masterTracker.shuffleStatuses.get(20).foreach { shuffleStatus =>
        shuffleStatus.cachedSerializedBroadcast.destroy(true)
      }
      val err = intercept[SparkException] {
        MapOutputTracker.deserializeOutputStatuses[MapStatus](fetchedBytes, conf)
      }
      assert(err.getMessage.contains("Unable to deserialize broadcasted output statuses"))
    }
  }

  test("SPARK-32921: test new protocol changes fetching both Map and Merge status in single RPC") {
    val newConf = new SparkConf
    newConf.set(RPC_MESSAGE_MAX_SIZE, 1)
    newConf.set(RPC_ASK_TIMEOUT, "1") // Fail fast
    newConf.set(SHUFFLE_MAPOUTPUT_MIN_SIZE_FOR_BROADCAST, 10240L) // 10 KiB << 1MiB framesize
    newConf.set(PUSH_BASED_SHUFFLE_ENABLED, true)
    newConf.set(IS_TESTING, true)

    // needs TorrentBroadcast so need a SparkContext
    withSpark(new SparkContext("local", "MapOutputTrackerSuite", newConf)) { sc =>
      val masterTracker = sc.env.mapOutputTracker.asInstanceOf[MapOutputTrackerMaster]
      val rpcEnv = sc.env.rpcEnv
      val masterEndpoint = new MapOutputTrackerMasterEndpoint(rpcEnv, masterTracker, newConf)
      rpcEnv.stop(masterTracker.trackerEndpoint)
      rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME, masterEndpoint)
      val bitmap1 = new RoaringBitmap()
      bitmap1.add(1)

      masterTracker.registerShuffle(20, 100, MergeStatus.SHUFFLE_PUSH_DUMMY_NUM_REDUCES)
      (0 until 100).foreach { i =>
        masterTracker.registerMapOutput(20, i, new CompressedMapStatus(
          BlockManagerId("999", "mps", 1000), Array.fill[Long](4000000)(0), 5))
      }
      masterTracker.registerMergeResult(20, 0, MergeStatus(BlockManagerId("999", "mps", 1000), 0,
        bitmap1, 1000L))

      val mapWorkerRpcEnv = createRpcEnv("spark-worker", "localhost", 0, new SecurityManager(conf))
      val mapWorkerTracker = new MapOutputTrackerWorker(conf)
      mapWorkerTracker.trackerEndpoint =
        mapWorkerRpcEnv.setupEndpointRef(rpcEnv.address, MapOutputTracker.ENDPOINT_NAME)

      val fetchedBytes = mapWorkerTracker.trackerEndpoint
        .askSync[(Array[Byte], Array[Byte])](GetMapAndMergeResultStatuses(20))
      assert(masterTracker.getNumAvailableMergeResults(20) == 1)
      assert(masterTracker.getNumAvailableOutputs(20) == 100)

      val mapOutput =
        MapOutputTracker.deserializeOutputStatuses[MapStatus](fetchedBytes._1, newConf)
      val mergeOutput =
        MapOutputTracker.deserializeOutputStatuses[MergeStatus](fetchedBytes._2, newConf)
      assert(mapOutput.length == 100)
      assert(mergeOutput.length == 1)
      mapWorkerTracker.stop()
      masterTracker.stop()
    }
  }

  test("SPARK-32921: unregister merge result if it is present and contains the map Id") {
    val rpcEnv = createRpcEnv("test")
    val tracker = newTrackerMaster()
    tracker.trackerEndpoint = rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME,
      new MapOutputTrackerMasterEndpoint(rpcEnv, tracker, conf))
    tracker.registerShuffle(10, 4, 2)
    assert(tracker.containsShuffle(10))
    val bitmap1 = new RoaringBitmap()
    bitmap1.add(0)
    bitmap1.add(1)
    tracker.registerMergeResult(10, 0, MergeStatus(BlockManagerId("a", "hostA", 1000), 0,
      bitmap1, 1000L))

    val bitmap2 = new RoaringBitmap()
    bitmap2.add(5)
    bitmap2.add(6)
    tracker.registerMergeResult(10, 1, MergeStatus(BlockManagerId("b", "hostB", 1000), 0,
      bitmap2, 1000L))
    assert(tracker.getNumAvailableMergeResults(10) == 2)
    tracker.unregisterMergeResult(10, 0, BlockManagerId("a", "hostA", 1000), Option(0))
    assert(tracker.getNumAvailableMergeResults(10) == 1)
    tracker.unregisterMergeResult(10, 1, BlockManagerId("b", "hostB", 1000), Option(1))
    assert(tracker.getNumAvailableMergeResults(10) == 1)
    tracker.unregisterMergeResult(10, 1, BlockManagerId("b", "hostB", 1000), Option(5))
    assert(tracker.getNumAvailableMergeResults(10) == 0)
    tracker.stop()
    rpcEnv.shutdown()
  }
}


text:
var express = require('express');
var app = express();
app.use(express.static(__dirname + '/public'));
app.use(express.static(__dirname + '/src'));
app.use(express.static(__dirname + '/bower_components'));
app.listen(3000);

app.get('/gallery/:index', function(req, res) {
	res.end('Slide ' + req.params.index + " from server");
});


text:
\documentclass{article}

\usepackage[T1]{fontenc}
\usepackage[osf]{libertine}
\usepackage[scaled=0.8]{beramono}
\usepackage[margin=1.5in]{geometry}
\usepackage{url}
\usepackage{booktabs}
\usepackage{amsmath}
\usepackage{amsfonts}
\usepackage{nicefrac}
\usepackage{microtype}
\usepackage{bm}

\usepackage{sectsty}
\sectionfont{\large}
\subsectionfont{\normalsize}

\usepackage{titlesec}
\titlespacing{\section}{0pt}{10pt plus 2pt minus 2pt}{0pt plus 2pt minus 0pt}
\titlespacing{\subsection}{0pt}{5pt plus 2pt minus 2pt}{0pt plus 2pt minus 0pt}

\setlength{\parindent}{0pt}
\setlength{\parskip}{1ex}

\newcommand{\acro}[1]{\textsc{\MakeLowercase{#1}}}
\newcommand{\given}{\mid}
\newcommand{\mc}[1]{\mathcal{#1}}
\newcommand{\data}{\mc{D}}
\newcommand{\model}{\mc{M}}
\newcommand{\trans}{^\top}
\newcommand{\intd}[1]{\,\mathrm{d}{#1}}
\newcommand{\mat}[1]{\bm{\mathrm{#1}}}
\renewcommand{\vec}[1]{\bm{\mathrm{#1}}}

\begin{document}

{\large \textbf{CSE 515T (Spring 2019) Assignment 3}} \\
Due Wednesday, 10 April 2019 \\

\begin{enumerate}
\item
  (Sampling from a multivariate Gaussian distribution).  Assume you can sample
  from a univariate standard normal distribution $\mc{N}(0, 1^2)$.  In the last
  assignment, this allowed you to sample trivially from a multivariate standard
  normal distribution $\mc{N}(\vec{0}, \mat{I})$.  Use the closure of the
  Gaussian distribution to affine transformations to derive a procedure to
  sample from an arbitrary multivariate Gaussian distribution
  \[
    p(\vec{x}) = \mc{N}(\vec{x}; \vec{\mu}, \mat{\Sigma})
  \]
  using an oracle that can draw a sample from the univariate standard normal
  distribution.

  Hint: the \emph{Cholesky decomposition} may be helpful.
\end{enumerate}

\clearpage
Now we will consider a series of questions relating to an application of
Bayesian inference to numerical analysis, specifically quadrature.

We are going to consider the function
\[f(x) = \exp(-x^2)\]
and its definite integral
\[Z = \int_{-\infty}^\infty f(x) \intd x.\]
The function $f$ has no elementary antiderivative, so the calculation of $Z$ is
not straightforward. There is a famous method for computing $Z$ with the trick
of considering $Z^2$ instead, rewriting the resulting 2$d$ integral in polar
coordinates, and making a convenient substitution.\footnote{This is commonly
  credited to Gauss, but the idea goes back at least to Poisson.} If you haven't
seen this, it's beautiful and worth checking out. The result is
\[Z = \sqrt{\pi}.\]

We will consider modeling $f$ with a Gaussian process prior distribution:
\[p(f) = \mc{GP}(f; \mu, K),\]
and conditioning on the following set of data $\data = (\vec{x}, \vec{y})$:
\begin{align*}
  \vec{x} &= [-2.5, -1.5, -0.5, 0.5, 1.5, 2.5]\trans; \\
  \vec{y} &= \exp(-\vec{x}^2) \\
          &= [0.0019305, 0.1054, 0.7788, 0.7788, 0.1054, 0.0019305]\trans.
\end{align*}
We will fix the prior mean function $\mu$ to be identically zero; $\mu(x) = 0$.

\begin{enumerate}
\setcounter{enumi}{1}
\item
  First, let us consider the question of model, specifically kernel, selection.

  Consider the following three choices for the covariance function $K$:
  \begin{align*}
    K_1(x, x') &= \exp\bigl(-\lvert x - x' \rvert^2\bigr) \\
    K_2(x, x') &= \exp\bigl(-\lvert x - x' \rvert\bigr) \\
    K_3(x, x') &= \bigl(1 + \sqrt{3} \lvert x - x' \rvert\bigr)
                  \exp\bigl(-\sqrt{3} \lvert x - x' \rvert\bigr).
  \end{align*}
  Note that I am not parameterizing any of these kernels; please consider them
  to be fixed.

  Each kernel defines a Gaussian process model for the data in a natural way:
  \[p(f \given \model_i) = \mc{GP}(f; \mu, K_i).\]
  Consider a uniform prior distribution over these models:
  \[\Pr(\model_i) = \nicefrac{1}{3} \qquad i = 1, 2, 3.\]

  \begin{enumerate}
  \item
    Compute the log marginal likelihood for each model given the data $\data$
    above.
  \item
    Compute the model posterior $\Pr(\model \given \data)$. Is there strong
    evidence for one model being the best?
  \end{enumerate}

\item
  Now we will consider integration.

  Perform Bayesian quadrature to estimate the definite integral $\int_{-5}^5
  f(x) \intd{x}$, using the model $\model_1$ from question 1.  What is the
  predictive mean and standard deviation, $p(Z \given \data, \model_1)$?  (You
  may give a numeric answer.)  How does this compare with the true answer?

\end{enumerate}

\end{document}


text:
<?xml version="1.0" encoding="UTF-8"?>
<jhove xmlns="http://hul.harvard.edu/ois/xml/ns/jhove" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" name="epubcheck" release="4.0.1-SNAPSHOT" date="2015-10-13">
 <date>2015-10-14T01:32:37+02:00</date>
 <repInfo uri="Epub3_marked_v2.epub">
  <created>2015-10-14T00:39:52Z</created>
  <lastModified>2012-10-10T12:00:00Z</lastModified>
  <format>application/epub+zip</format>
  <version>2.0.1</version>
  <status>Not well-formed</status>
  <messages>
   <message severity="error" subMessage="RSC-005">RSC-005, ERROR, [Error while parsing file 'attribute "property" not allowed here; expected attribute "content", "id", "name", "scheme" or "xml:lang"'.], OPS/content.opf (7-39)</message>
   <message severity="error" subMessage="RSC-005">RSC-005, ERROR, [Error while parsing file 'element "meta" missing required attributes "content" and "name"'.], OPS/content.opf (7-39)</message>
   <message severity="error" subMessage="RSC-005">RSC-005, ERROR, [Error while parsing file 'text not allowed here; expected the element end-tag'.], OPS/content.opf (7-61)</message>
   <message severity="error" subMessage="RSC-005">RSC-005, ERROR, [Error while parsing file 'attribute "properties" not allowed here; expected attribute "fallback", "fallback-style", "href", "media-type", "required-modules" or "required-namespace"'.], OPS/content.opf (12-90)</message>
   <message severity="error" subMessage="HTM-004">HTM-004, ERROR, [Irregular DOCTYPE: found '', expected '<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.1//EN"
    "http://www.w3.org/TR/xhtml11/DTD/xhtml11.dtd">'.], OPS/toc.xhtml</message>
   <message severity="error" subMessage="RSC-005">RSC-005, ERROR, [Error while parsing file 'element "nav" not allowed anywhere; expected element "address", "blockquote", "del", "div", "dl", "h1", "h2", "h3", "h4", "h5", "h6", "hr", "ins", "noscript", "ns:svg", "ol", "p", "pre", "script", "table" or "ul" (with xmlns:ns="http://www.w3.org/2000/svg")'.], OPS/toc.xhtml (9-22)</message>
   <message severity="error" subMessage="RSC-005">RSC-005, ERROR, [Error while parsing file 'element "nav" not allowed anywhere; expected element "address", "blockquote", "del", "div", "dl", "h1", "h2", "h3", "h4", "h5", "h6", "hr", "ins", "noscript", "ns:svg", "ol", "p", "pre", "script", "table" or "ul" (with xmlns:ns="http://www.w3.org/2000/svg")'.], OPS/toc.xhtml (12-10)</message>
   <message severity="error" subMessage="RSC-005">RSC-005, ERROR, [Error while parsing file 'element "body" incomplete; expected element "address", "blockquote", "del", "div", "dl", "h1", "h2", "h3", "h4", "h5", "h6", "hr", "ins", "noscript", "ns:svg", "ol", "p", "pre", "script", "table" or "ul" (with xmlns:ns="http://www.w3.org/2000/svg")'.], OPS/toc.xhtml (14-8)</message>
   <message severity="info" subMessage="HTM-010">HTM-010, HINT, [Namespace uri 'http://www.daisy.org/z3986/2005/ncx/' was found.], OPS/toc.ncx (2-68)</message>
  </messages>
  <mimeType>application/epub+zip</mimeType>
  <properties>
   <property>
    <name>CharacterCount</name>
    <values arity="Scalar" type="Long">
     <value>223</value>
    </values>
   </property>
   <property>
    <name>Language</name>
    <values arity="Scalar" type="String">
     <value>en</value>
    </values>
   </property>
   <property>
    <name>Info</name>
    <values arity="List" type="Property">
     <property>
      <name>Identifier</name>
      <values arity="Scalar" type="String">
       <value>000000000000000000</value>
      </values>
     </property>
     <property>
      <name>CreationDate</name>
      <values arity="Scalar" type="Date">
       <value>2015-10-14T00:39:52Z</value>
      </values>
     </property>
     <property>
      <name>ModDate</name>
      <values arity="Scalar" type="Date">
       <value>2012-10-10T12:00:00Z</value>
      </values>
     </property>
     <property>
      <name>Title</name>
      <values arity="Scalar" type="String">
       <value>Missing TOC Sample Book</value>
      </values>
     </property>
    </values>
   </property>
   <property>
      <name>MediaTypes</name>
      <values arity="Array" type="String">
         <value>application/x-dtbncx+xml</value>
         <value>application/xhtml+xml</value>
      </values>
   </property>
  </properties>
 </repInfo>
</jhove>


text:
<?php

/* TestBlogBundle:security:login.html.twig */
class __TwigTemplate_ef581f3afb2ceb1dc21bcc2952156b0fdb2c0c26db18393153a6e5e1abf2912b extends Twig_Template
{
    public function __construct(Twig_Environment $env)
    {
        parent::__construct($env);

        $this->parent = false;

        $this->blocks = array(
        );
    }

    protected function doDisplay(array $context, array $blocks = array())
    {
        $__internal_43931f357e2aa2a3eae00c94c48477e3801c31dd3629ec60fbcfca7d6dd33e05 = $this->env->getExtension("native_profiler");
        $__internal_43931f357e2aa2a3eae00c94c48477e3801c31dd3629ec60fbcfca7d6dd33e05->enter($__internal_43931f357e2aa2a3eae00c94c48477e3801c31dd3629ec60fbcfca7d6dd33e05_prof = new Twig_Profiler_Profile($this->getTemplateName(), "template", "TestBlogBundle:security:login.html.twig"));

        // line 1
        if ((isset($context["error"]) ? $context["error"] : $this->getContext($context, "error"))) {
            // line 2
            echo "
";
            echo twig_escape_filter($this->env, $this->env->getExtension('translator')->trans($this->getAttribute((isset($context["error"]) ? $context["error"] : $this->getContext($context, "error")), "messageKey", array()), $this->getAttribute((isset($context["error"]) ? $context["error"] : $this->getContext($context, "error")), "messageData", array()), "security"), "html", null, true);
            echo "

";
        }
        // line 4
        echo "
<form action=\"";
        // line 5
        echo $this->env->getExtension('routing')->getPath("login_check");
        echo "\" method=\"post\" class=\"form\">
    Username:
    <input type=\"text\" id=\"username\" name=\"_username\" value=\"";
        // line 7
        echo twig_escape_filter($this->env, (isset($context["last_username"]) ? $context["last_username"] : $this->getContext($context, "last_username")), "html", null, true);
        echo "\" />

    Password:
    <input type=\"password\" id=\"password\" name=\"_password\" />

    <input type=\"hidden\" name=\"_target_path\" value=\"/\" />

    <button type=\"submit\">login</button>
</form>";

        $__internal_43931f357e2aa2a3eae00c94c48477e3801c31dd3629ec60fbcfca7d6dd33e05->leave($__internal_43931f357e2aa2a3eae00c94c48477e3801c31dd3629ec60fbcfca7d6dd33e05_prof);

    }

    public function getTemplateName()
    {
        return "TestBlogBundle:security:login.html.twig";
    }

    public function isTraitable()
    {
        return false;
    }

    public function getDebugInfo()
    {
        return array (  38 => 7,  33 => 5,  30 => 4,  24 => 2,  22 => 1,);
    }
}


text:
<menu xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    xmlns:tools="http://schemas.android.com/tools"
    tools:context=".MainActivity" >

    <item android:id="@+id/action_write"
        android:icon="@android:drawable/ic_menu_edit"
        android:title="Write Article"
        android:orderInCategory="100"
        app:showAsAction="always" />
</menu>


text:
require 'spec_helper'

describe 'variable_contains_upcase' do
  let(:msg) { 'variable contains a capital letter' }

  context 'a variable containing a capital' do
    let(:code) { '$fOobar' }

    it 'should only detect a single problem' do
      expect(problems).to have(1).problem
    end

    it 'should create a warning' do
      expect(problems).to contain_warning(msg).on_line(1).in_column(1)
    end
  end

  context 'variable containing a capital' do
    let(:code) { '" $fOobar"' }

    it 'should only detect a single problem' do
      expect(problems).to have(1).problem
    end

    it 'should create a warning' do
      expect(problems).to contain_warning(msg).on_line(1).in_column(3)
    end
  end

  context 'with fix enabled' do
    before do
      PuppetLint.configuration.fix = true
    end

    after do
      PuppetLint.configuration.fix = false
    end

    context 'fix variable containing a capital' do
      let(:code) { '" $fOobar"' }

      it 'should only detect a single problem' do
        expect(problems).to have(1).problem
      end

      it 'should fix the manifest' do
        expect(problems).to contain_fixed(msg).on_line(1).in_column(3)
      end

      it 'should downcase the variable name' do
        expect(manifest).to eq('" $foobar"')
      end
    end

    context 'a hash with an upcase key should not fail' do
      let(:code) { ' $myhash["Foo"]' }

      it 'should not detect a single problem' do
        expect(problems).to have(0).problem
      end
    end
    context 'a hash with an upcase key should not fail' do
      let(:code) { ' $myhash["Foo"]["bAr"]' }

      it 'should not detect a single problem' do
        expect(problems).to have(0).problem
      end
    end
    context 'a hash variable with an upcase key should fail' do
      let(:code) { ' $myHash["foo"]["bar"]' }

      it 'should detect a single problem' do
        expect(problems).to have(1).problem
      end
    end
  end
end


text:
<?php
function loadCallback($file, $parser)
{
    $paths = array();
    foreach ($parser->extensions as $extensionName) {
        $namespace = ucwords(preg_replace('/[^0-9a-z]+/', '_', strtolower($extensionName)));
        $extensionPath = './' . $namespace . '/' . $namespace . '.php';
        if (file_exists($extensionPath)) {
            require_once($extensionPath);
            $hook = $namespace . '::resolveExtensionPath';
            $returnPath = call_user_func($hook, $file, $parser);
            if (!empty($returnPath)) {
                $paths[] = $returnPath;
            }

        }
    }
    return $paths;
}

function getFunctions($extensions)
{
    $output = array();
    if (!empty($extensions)) {
        foreach ($extensions as $extension) {
            $name = explode('/', $extension, 2);
            $namespace = ucwords(preg_replace('/[^0-9a-z]+/', '_', strtolower(array_shift($name))));
            $extensionPath = './' . $namespace . '/' . $namespace . '.php';
            if (file_exists(
                $extensionPath
            )
            ) {
                require_once($extensionPath);
                $namespace = $namespace . '::';
                $function = 'getFunctions';
                $output = array_merge($output, call_user_func($namespace . $function, $namespace));
            }
        }
    }

    return $output;
}

$file = 'example.scss';
$path = '../';
$library = $path . '/SassParser.php';

if ($path && file_exists($library)) {

    try {
        require_once ($library);

        $options = array(
            'style' => 'expanded',
            'cache' => false,
            'syntax' => 'scss',
            'debug' => false,
            'debug_info' => false,
            'load_path_functions' => array('loadCallback'),
            'load_paths' => array(dirname($file)),
            'functions' => getFunctions(array('Compass','Own')),
            'extensions' => array('Compass','Own')
        );
        // Execute the compiler.
        $parser = new SassParser($options);
        print $parser->toCss($file);
    } catch (Exception $e) {
        print "body::before {
          display: block;
          padding: 5px;
          white-space: pre;
          font-family: monospace;
          font-size: 8pt;
          line-height: 17px;
          overflow: hidden;
          content: '" . $e->getMessage() . "';
        }";
    }
} else {
    echo 'Path or library are wrong';
}


text:
package org.apache.camel.component.kubernetes.deployments;

import org.apache.camel.Category;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.component.kubernetes.AbstractKubernetesEndpoint;
import org.apache.camel.component.kubernetes.KubernetesConfiguration;
import org.apache.camel.spi.UriEndpoint;

/**
 * Perform operations on Kubernetes Deployments and get notified on Deployment changes.
 */
@UriEndpoint(firstVersion = "2.20.0", scheme = "kubernetes-deployments", title = "Kubernetes Deployments",
             syntax = "kubernetes-deployments:masterUrl", category = { Category.CONTAINER, Category.CLOUD, Category.PAAS })
public class KubernetesDeploymentsEndpoint extends AbstractKubernetesEndpoint {

    public KubernetesDeploymentsEndpoint(String uri, KubernetesDeploymentsComponent component, KubernetesConfiguration config) {
        super(uri, component, config);
    }

    @Override
    public Producer createProducer() throws Exception {
        return new KubernetesDeploymentsProducer(this);
    }

    @Override
    public Consumer createConsumer(Processor processor) throws Exception {
        Consumer consumer = new KubernetesDeploymentsConsumer(this, processor);
        configureConsumer(consumer);
        return consumer;
    }

}


text:
Useful scripts for writting ClusterFuzz fuzzers.

https://www.chromium.org/Home/chromium-security/bugs/developing-fuzzers-for-clusterfuzz


text:
For more details on creating an App Service Certificate see [How to Create an App Service Certificate](https://azure.microsoft.com/en-us/documentation/articles/web-sites-purchase-ssl-web-site/).

In order to deploy this template, you need to have the following resources:

1. A Key Vault (specified in 'existingKeyVaultId' parameter)

2. An App Service App(specified in 'existingAppName' parameter)

3. An App Service Domain (specified in 'rootHostName' parameter)

By default, 'Microsoft.CertificateRegistration' and 'Microsoft.Web' RPs don't have access to the Key Vault specified in the template hence you need to authorize these RPs by executing
the following PowerShell commands before deploying the template:

Login-AzureRmAccount

Set-AzureRmContext -SubscriptionId AZURE_SUBSCRIPTION_ID

Set-AzureRmKeyVaultAccessPolicy -VaultName KEY_VAULT_NAME -ServicePrincipalName f3c21649-0979-4721-ac85-b0216b2cf413 -PermissionsToSecrets get,set,delete

Set-AzureRmKeyVaultAccessPolicy -VaultName KEY_VAULT_NAME -ServicePrincipalName abfa0a7c-a6b6-4736-8310-5855508787cd -PermissionsToSecrets get

ServicePrincipalName parameter represents these RPs in user tenant and will remain same for all Azure subscriptions. This is a onetime operation. Once you have a configured a Key Vault property,
you can use it to store as many App Service Certificates as you want without executing these PowerShell commands again. You can go through the Key Vault documentation for more information:

https://azure.microsoft.com/en-us/documentation/articles/key-vault-get-started/

The Web App and domain resources need to be in the same resource group. The Web App should have 'rootHostName' and www subdomain assigned as custom domains.

https://azure.microsoft.com/en-us/documentation/articles/custom-dns-web-site-buydomains-web-app/

https://azure.microsoft.com/en-us/documentation/articles/web-sites-custom-domain-name/


text: export { FindOneAndUpdate } from './findOneAndUpdate';


text:
<md-content class="md-whiteframe-z3">

Login

        <md-text-float flex label="Username" ng-model="username"></md-text-float>
        <md-text-float flex label="Password" ng-model="password" type="password"></md-text-float>

        <md-button class="md-accent md-raised md-hue-3" ng-click="loginUser()">Login</md-button>

</md-content>


text:
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">

<title>JsonUnivariateAggregator (Edgent v0.4.0)</title>
<meta name="date" content="2016-07-26">
<link rel="stylesheet" type="text/css" href="../../../../../../stylesheet.css" title="Style">

JavaScript is disabled on your browser.

Skip navigation links

Overview

Package

Class

Use

Tree

Deprecated

Index

Help

Prev Class

Next Class

Frames

No Frames

All Classes

Summary:

Nested |

Field |

Constr |

Method

Detail:

Field |

Constr |

Method

org.apache.edgent.analytics.math3.json

Interface JsonUnivariateAggregator

All Known Implementing Classes:

JsonStorelessStatistic

public interface JsonUnivariateAggregator

Univariate aggregator for JSON tuples.
 This is the runtime implementation interface
 of JsonUnivariateAggregate defined aggregate.

Method Summary

All Methods Instance Methods Abstract Methods

Modifier and Type
Method and Description

void
clear(com.google.gson.JsonElement partitionKey,
     int n)

Clear the aggregator to prepare for a new aggregation.

void
increment(double value)

Add a value to the aggregation.

void
result(com.google.gson.JsonElement partitionKey,
      com.google.gson.JsonObject result)

Place the result of the aggregation into the result
 object.

Method Detail

clear

void clear(com.google.gson.JsonElement partitionKey,
           int n)

Clear the aggregator to prepare for a new aggregation.

Parameters:

partitionKey - Partition key.

n - Number of tuples to be aggregated.

increment

void increment(double value)

Add a value to the aggregation.

Parameters:

value - Value to be added.

result

void result(com.google.gson.JsonElement partitionKey,
            com.google.gson.JsonObject result)

Place the result of the aggregation into the result
 object. The key for the result must be
 JsonUnivariateAggregate.name() for the corresponding
 aggregate. The value of the aggregation may be a primitive value
 such as a double or any valid JSON element.

Parameters:

partitionKey - Partition key.

result - JSON object holding the result.

Skip navigation links

Overview

Package

Class

Use

Tree

Deprecated

Index

Help

Apache Edgent (incubating)

Prev Class

Next Class

Frames

No Frames

All Classes

Summary:

Nested |

Field |

Constr |

Method

Detail:

Field |

Constr |

Method

Copyright © 2016 The Apache Software Foundation. All Rights Reserved - 54ea8ac-20160726-1240