TaxonomyServiceHandler.java

package com.amplitude.taxonomy;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.math.RandomUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.thrift.TException;
import org.leadpony.justify.api.JsonSchema;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

import com.amplitude.core.util.Config;
import com.amplitude.core.util.JsonUtils;
import com.amplitude.core.util.MetricsWrapper;
import com.amplitude.core.util.concurrent.DaemonThreadFactory;
import com.amplitude.core.util.concurrent.MetricsTrackedThreadPoolExecutor;
import com.amplitude.dynconfv2.client.Dynconfv2;
import com.amplitude.gigatron.model.AmplitudeEventV13;
import com.amplitude.ingestion.dbs.AccountsService;
import com.amplitude.ingestion.dbs.FeatureFlagSet;
import com.amplitude.ingestion.dbs.FeatureFlagsService;
import com.amplitude.stargate.observe.StargateObserve;
import com.amplitude.taxonomy.TaxonomyValidator.SchemaValidationType;
import com.amplitude.taxonomy.model.AppIdToUserPropJsonSchema;
import com.amplitude.taxonomy.model.coredata.PropertySchema;
import com.amplitude.taxonomy.model.coredata.TaxonomyWhitelistSettings;
import com.amplitude.taxonomy.model.coredata.UserPropertyMetadata;
import com.amplitude.taxonomy.model.coredata.UserPropertySchema;
import com.amplitude.taxonomy.model.thrift.*;
import com.amplitude.taxonomy.thrift.TTaxonomyService.Iface;

public class TaxonomyServiceHandler implements Iface {
    private static final Logger LOGGER = LogManager.getLogger();
    // Orion-related feature flag that more or less marks that the org is ready for ingestion shadowing
    protected static final String STARGATE_ORION_SOURCE_OF_TRUTH_FEATURE_FLAG = "stargate-orion-as-source-of-truth";

    // Orion-related feature flag to control a two-phased rollout for taxonomy validation where we first
    // validate non-ingestion-related changes to reduce the potential blast radius
    protected static final String STARGATE_ORION_INGESTION_FEATURE_FLAG = "stargate-orion-ingestion";

    protected static final String CONSOLIDATED_TAXONOMY_DYNCONF_SHADOWING_ENABLED_KEY = "taxonomy.consolidatedValidator.enabled";

    public static final String CONSOLIDATED_TAXONOMY_ROLLOUT_DYNCONF_KEY = "taxonomy.consolidatedValidator.rolloutEnabled.orgs";

    public static final String CONSOLIDATED_TAXONOMY_SHADOW_DYNCONF_KEY = "taxonomy.consolidatedValidator.shadowingEnabled.orgs";

    protected static final String LOG_CONSOLIDATED_SHADOW_MISMATCH_DYNCONF_KEY = "taxonomy.logConsolidatedShadowMismatch";

    protected static final String APPS_TO_LOG_EVENT_CONSOLIDATED_SHADOW_MISMATCH_DYNCONF_KEY = "taxonomy.logConsolidatedShadowMismatch.apps";

    protected static final String LEGACY_VERSION_TAG = "version:legacy";
    protected static final String CONSOLIDATED_VERSION_TAG = "version:consolidated";

    protected static final double RELATIVE_TOLERANCE = 1e-3;
    protected static final double ABSOLUTE_TOLERANCE = 1e-10;

    private static final int DEFAULT_COMPARE_THREADS = 50;
    private static final int DEFAULT_COMPARE_QUEUE_MAX_SIZE = 500;

    private final TaxonomyValidator schemaValidator;
    private final StargateObserve stargateObserve;

    private final ConsolidatedTaxonomyValidator consolidatedValidator;
    private final FeatureFlagsService featureFlagsService;
    private final AccountsService accountsService;
    private final ExecutorService comparisonExecutor;

    RejectedExecutionHandler rejectedComparisonHandler = new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            MetricsWrapper.incrementCounter("taxonomy.shadowComparison.rejected");
        }
    };

    public TaxonomyServiceHandler(
            TaxonomyValidator schemaValidator,
            StargateObserve stargateObserve,
            ConsolidatedTaxonomyValidator consolidatedValidator,
            FeatureFlagsService featureFlagsService,
            AccountsService accountsService) {
        this.schemaValidator = schemaValidator;
        this.stargateObserve = stargateObserve;
        this.consolidatedValidator = consolidatedValidator;
        this.featureFlagsService = featureFlagsService;
        this.accountsService = accountsService;

        // Initialize the comparison executor, which should additionally limit the comparisons
        int comparisonThreadsCount = Dynconfv2.instance().getInt("taxonomy.comparison.threads",
                DEFAULT_COMPARE_THREADS);
        int comparisonQueueMaxSize = Dynconfv2.instance().getInt("taxonomy.comparison.queueMaxSize",
                DEFAULT_COMPARE_QUEUE_MAX_SIZE);
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(comparisonQueueMaxSize);
        this.comparisonExecutor = new MetricsTrackedThreadPoolExecutor(comparisonThreadsCount, comparisonQueueMaxSize,
                60,
                TimeUnit.SECONDS, workQueue, new DaemonThreadFactory("taxononmy-shadow-compare"),
                rejectedComparisonHandler, "taxonomyShadowCompare");
    }

    private Map<Integer, TWhitelistSettings> buildThriftWhitelistSettings(
            Map<Integer, TaxonomyWhitelistSettings> whitelistSettings) {
        Map<Integer, TWhitelistSettings> thriftWhitelistSettings = Maps.newHashMap();
        for (Map.Entry<Integer, TaxonomyWhitelistSettings> entry : whitelistSettings.entrySet()) {
            thriftWhitelistSettings.put(entry.getKey(), entry.getValue().toTWhitelistSettings());
        }
        return thriftWhitelistSettings;
    }

    private Map<Integer, List<TUserPropertyMetadata>> buildThriftUserPropSchema(
            Map<Integer, Collection<UserPropertyMetadata>> appIdToUserPropSchema) {
        Map<Integer, List<TUserPropertyMetadata>> thriftUserPropSchemas = Maps.newHashMap();
        if (appIdToUserPropSchema == null)
            return thriftUserPropSchemas;

        for (Map.Entry<Integer, Collection<UserPropertyMetadata>> entry : appIdToUserPropSchema.entrySet()) {
            List<TUserPropertyMetadata> listOfMetadata = entry.getValue().stream()
                    .map(UserPropertyMetadata::toTUserPropertyMetadata)
                    .collect(Collectors.toList());
            thriftUserPropSchemas.put(entry.getKey(), listOfMetadata);
        }
        return thriftUserPropSchemas;
    }

    private Map<Integer, String> buildThriftUserPropJsonSchema(
            AppIdToUserPropJsonSchema appIdToUserPropJsonSchema) {
        Map<Integer, String> thriftUserPropSchemas = Maps.newHashMap();
        if (appIdToUserPropJsonSchema == null) {
            return thriftUserPropSchemas;
        }
        for (Map.Entry<Integer, JsonSchema> entry : appIdToUserPropJsonSchema.getAppIdToJsonSchema().entrySet()) {
            thriftUserPropSchemas.put(entry.getKey(), entry.getValue().toString());
        }
        return thriftUserPropSchemas;
    }

    private List<String> buildThriftEventsOrProperties(List<JSONObject> eventsOrProperiesObjects) {
        // Didn't see a nice way for fast json to make a null, but didn't look around very hard. Opted to use the string
        // literal of "null" which is the json serialized form of json null
        return eventsOrProperiesObjects.stream()
                // Its critical that we preserve null values to maintain parity with the old system.
                .map(e -> e == null ? "null" : JSONObject.toJSONString(e, SerializerFeature.WriteMapNullValue))
                .collect(Collectors.toList());
    }

    @VisibleForTesting
    static List<TEventValidationResult> thriftResponseFromConsolidatedTaxonomyResult(TaxonomyServiceHandler handler,
            ValidationResponse consolidatedResponse) {
        List<TEventValidationResult> result = new ArrayList<>();

        TSchemaValidation tSchemaValidation = handler.thriftResponseFromValidationResult(consolidatedResponse);

        TEventValidationResult tEventValidationResult = new TEventValidationResult(
                tSchemaValidation.sharedSchemaManagerId, tSchemaValidation.events,
                tSchemaValidation.appIdToWhitelistSettings, tSchemaValidation.appIdToUserPropSchema);
        tEventValidationResult
                .setAppIdToDataplaneUserPropChecksum(tSchemaValidation.appIdToDataplaneUserPropChecksum);
        tEventValidationResult.setAppIdToDataplaneUserPropSchema(tEventValidationResult.appIdToDataplaneUserPropSchema);
        result.add(tEventValidationResult);
        return result;
    }

    @VisibleForTesting
    TSchemaValidation thriftResponseFromValidationResult(ValidationResponse validationResult) {
        TSchemaValidation tSchemaValidation = new TSchemaValidation();
        tSchemaValidation
                .setAppIdToWhitelistSettings(buildThriftWhitelistSettings(validationResult.taxonomyWhitelistSettings));
        tSchemaValidation.setAppIdToUserPropSchema(buildThriftUserPropSchema(validationResult.appIdToUserPropSchema));
        tSchemaValidation.setAppIdToUserPropSchemaChecksum(validationResult.appIdToUserPropSchemaChecksum);
        // either eventsOrPropertiesObjects or eventsOrPropertiesJsonStrings should ever be set.
        // if eventsOrProperiesObjects is set and eventsOrPropertiesJsonStrings is not, then set events to
        // eventsOrProperiesObjects
        // if eventsOrPropertiesJsonStrings is set and eventsOrProperiesObjects is not, then set events to
        // eventsOrPropertiesJsonStrings
        // if neither is set, then set events to an empty list
        // if both are set, then set to eventsOrPropertiesObjects FOR NOW & emit a metric because its unexpected
        boolean isOriginalListEmpty = CollectionUtils.isEmpty(validationResult.eventsOrProperiesObjects);
        boolean isConsolidatedListEmpty = CollectionUtils.isEmpty(validationResult.eventsOrPropertiesJsonStrings);
        if (isOriginalListEmpty && isConsolidatedListEmpty) {
            tSchemaValidation.setEvents(new ArrayList<>());
        } else if (!isOriginalListEmpty && isConsolidatedListEmpty) {
            tSchemaValidation.setEvents(buildThriftEventsOrProperties(validationResult.eventsOrProperiesObjects));
        } else if (isOriginalListEmpty && !isConsolidatedListEmpty) {
            // Consolidated validator responded with this
            tSchemaValidation.setEvents(validationResult.eventsOrPropertiesJsonStrings);
        } else {
            // This is unexpected, but we'll set it to the original list for now to maintain the peace
            tSchemaValidation.setEvents(buildThriftEventsOrProperties(validationResult.eventsOrProperiesObjects));
            MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.rollout.issue", 1,
                    "issue:bothListsSet");
        }

        tSchemaValidation.setAppIdToDataplaneUserPropSchema(
                buildThriftUserPropJsonSchema(validationResult.appIdToDataplaneUserPropSchema));
        tSchemaValidation.setAppIdToDataplaneUserPropChecksum(validationResult.appIdToDataplaneUserPropChecksum);
        return tSchemaValidation;
    }

    private boolean isStargateOrionAsSourceOfTruthEnabledForOrg(int orgId) {
        FeatureFlagSet featureFlagSet = featureFlagsService.getFeaturesForOrg(orgId);
        return featureFlagSet.hasFeature(STARGATE_ORION_SOURCE_OF_TRUTH_FEATURE_FLAG);
    }

    private boolean isStargateOrionIngestionEnabledForOrg(int orgId) {
        FeatureFlagSet featureFlagSet = featureFlagsService.getFeaturesForOrg(orgId);
        return featureFlagSet.hasFeature(STARGATE_ORION_INGESTION_FEATURE_FLAG);
    }

    private boolean isInShadowingSample() {
        double sampleRate = Dynconfv2.instance().getDouble("taxonomy.consolidatedValidator.shadowComparisonSampleRate",
                0.95);
        return RandomUtils.nextDouble() >= sampleRate;
    }

    private boolean shouldShadowConsolidatedValidator(int orgId) {
        return isInShadowingSample()
                && Dynconfv2.instance().getBoolean(CONSOLIDATED_TAXONOMY_DYNCONF_SHADOWING_ENABLED_KEY, false)
                && Dynconfv2.instance().isRolledOutForOrg(CONSOLIDATED_TAXONOMY_SHADOW_DYNCONF_KEY, orgId)
                && isStargateOrionAsSourceOfTruthEnabledForOrg(orgId);
    }

    private boolean doPropertiesMismatchAndLog(int appId, String eventType, String propertiesType,
            JSONObject oldVersionProperties, JSONObject consolidatedVersionProperties, JSONObject inputProperties) {
        JsonUtils.JSONEqualityResponse equalityResponse = JsonUtils.areJSONObjectsApproximatelyEqual(
                oldVersionProperties, consolidatedVersionProperties, RELATIVE_TOLERANCE, ABSOLUTE_TOLERANCE);
        if (!equalityResponse.areEqual()) {
            if (Dynconfv2.instance().getBoolean(LOG_CONSOLIDATED_SHADOW_MISMATCH_DYNCONF_KEY, false)) {
                LOGGER.info("Mismatched event: app: {}, event type: {}. propertiesType: {}. Reasons: {}", appId,
                        eventType,
                        propertiesType, equalityResponse.reasons());
                // Only log full properties for specific apps. keep this list to internal test apps to prevent logging
                // PII
                if (Dynconfv2.instance().isAppInList(APPS_TO_LOG_EVENT_CONSOLIDATED_SHADOW_MISMATCH_DYNCONF_KEY,
                        appId)) {
                    LOGGER.info("Input {} Properties: {}", propertiesType, inputProperties);
                    LOGGER.info("Original {} Properties: {}", propertiesType, oldVersionProperties);
                    LOGGER.info("Consolidated {} Properties: {}", propertiesType, consolidatedVersionProperties);
                }
            }
            String tag = String.format("issue:%sPropsMismatch", propertiesType);
            MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.eventMismatch", 1,
                    tag);
            return true;
        }
        return false;
    }

    /**
     * Helper for shadowing while we transition to the ConsolidatedValidator.
     * Since the ConsolidatedValidator will return different values for the checksums, and different types for the json strings,
     * it doesn't make sense to compare those. So we'll only compare events and whitelist settings.
     * @param validationResponse
     * @param consolidatedResponse
     * @return true if there is a mismatch in the actual properties
     */
    @VisibleForTesting
    protected boolean compareConsolidatedResponse(Integer appId, ValidationResponse validationResponse,
            ValidationResponse consolidatedResponse, List<JSONObject> inputEventPropertiesList,
            List<JSONObject> inputUserPropertiesList) {
        try {
            // compare the whitelist settings
            for (Map.Entry<Integer, TaxonomyWhitelistSettings> entry : validationResponse.taxonomyWhitelistSettings
                    .entrySet()) {
                compareWhiteListEntry(entry, consolidatedResponse.taxonomyWhitelistSettings,
                        "taxonomy.consolidatedValidator.whitelistMismatch");
            }

            if (validationResponse != null && consolidatedResponse == null ||
                    validationResponse == null && consolidatedResponse != null) {
                MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.eventMismatch", 1,
                        "issue:nullResponse");
                return true;
            }

            if (validationResponse.eventsOrProperiesObjects.size() != consolidatedResponse.eventsOrPropertiesJsonStrings
                    .size()) {
                MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.eventMismatch", 1,
                        "issue:diffPropSize");
                return true;
            }

            boolean anyMismatch = false;
            // Compare to the old response and log if there is a difference
            for (int i = 0; i < validationResponse.eventsOrProperiesObjects.size(); i++) {
                JSONObject eventJson = validationResponse.eventsOrProperiesObjects.get(i);
                String consolidatedEventStr = consolidatedResponse.eventsOrPropertiesJsonStrings.get(i);
                JSONObject inputEventProperties = inputEventPropertiesList.get(i);
                JSONObject inputUserProperties = inputUserPropertiesList.get(i);
                // The consolidated string will have its keys alphabetical, so its hard to compare to the alibaba
                // eventJson
                // Without doing this
                JSONObject consolidatedEvent = JsonUtils.parseJson(AmplitudeEventV13.getObjectMapper(),
                        consolidatedEventStr);
                if (eventJson == null && consolidatedEvent == null) {
                    MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.eventMismatch", 0, "issue:none");
                    continue;
                } else if (eventJson == null) {
                    MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.eventMismatch", 1,
                            "issue:nullEvent");
                    anyMismatch = true;
                    LOGGER.info("Mismatched event: app: {}, event type: {}. eventJson == null", appId,
                            consolidatedEvent.getString("event_type"));
                    continue;
                } else if (consolidatedEvent == null) {
                    MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.eventMismatch", 1,
                            "issue:nullConsolidatedEvent");
                    anyMismatch = true;
                    LOGGER.info("Mismatched event: app: {}, event type: {}. consolidatedEvent == null",
                            appId, eventJson.getString("event_type"));
                    continue;
                }

                JSONObject eventProps = eventJson.getJSONObject("event_properties");
                JSONObject consolidatedEventProps = consolidatedEvent.getJSONObject("event_properties");
                JSONObject userProps = eventJson.getJSONObject("user_properties");
                JSONObject consolidatedUserProps = consolidatedEvent.getJSONObject("user_properties");

                boolean eventPropsMismatch = doPropertiesMismatchAndLog(appId, eventJson.getString("event_type"),
                        "Event", eventProps, consolidatedEventProps, inputEventProperties);
                boolean userPropsMismatch = doPropertiesMismatchAndLog(appId, eventJson.getString("event_type"),
                        "User", userProps, consolidatedUserProps, inputUserProperties);
                boolean mismatch = userPropsMismatch || eventPropsMismatch;

                anyMismatch = anyMismatch || mismatch;
            }
            return anyMismatch;
        } catch (Exception e) {
            LOGGER.error("Encountered error in compareConsolidatedResponse validator for app: " + appId, e);
            MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.compareConsolidatedResponse.errors",
                    1);
            return true;
        }
    }

    /**
     * Helper for shadowing while we transition to the ConsolidatedValidator.
     * Since the ConsolidatedValidator will return different values for the checksums, and different types for the json strings,
     * it doesn't make sense to compare those. So we'll only compare events and whitelist settings.
     * @param batchIdentValidationResult - the result from the old validator
     * @param consolidatedResponse - the result from the new validator
     * @return true if there is any mismatch in the properties
     */
    @VisibleForTesting
    protected static boolean doesConsolidatedIdentifyResponseDiffer(Integer appId,
            TBatchIdentValidationResult batchIdentValidationResult,
            IdentifyValidationResponse consolidatedResponse) {
        try {
            if (batchIdentValidationResult == null && consolidatedResponse == null) {
                return false;
            }
            if (batchIdentValidationResult == null || consolidatedResponse == null) {
                MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.identify.eventMismatch", 1,
                        "issue:nullResponse");
                return true;
            }

            // compare the whitelist settings
            for (Map.Entry<Integer, TaxonomyWhitelistSettings> entry : consolidatedResponse.taxonomyWhitelistSettings
                    .entrySet()) {
                compareWhiteListEntry(entry, consolidatedResponse.taxonomyWhitelistSettings,
                        "taxonomy.consolidatedValidator.identify.whitelistMismatch");
            }

            // Next, compare the resulting user properties
            if (batchIdentValidationResult.userProperties.size() != consolidatedResponse.jsonPropertiesBatches.size()) {
                MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.identify.eventMismatch", 1,
                        "issue:batchSizeMismatch");
                return true;
            }

            boolean anyMismatch = false;
            // Compare to the old response and log if there is a difference
            for (int i = 0; i < batchIdentValidationResult.userProperties.size(); i++) {
                List<String> batch = batchIdentValidationResult.userProperties.get(i);
                List<String> consolidatedBatch = consolidatedResponse.jsonPropertiesBatches.get(i);
                if (batch.size() != consolidatedBatch.size()) {
                    MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.identify.eventMismatch", 1,
                            "issue:batchPropsSizeMismatch");
                    anyMismatch = true;
                    continue;
                }

                for (int j = 0; j < batch.size(); j++) {
                    String originalJsonStr = batch.get(j);
                    String consolidatedJsonStr = consolidatedBatch.get(j);
                    if (originalJsonStr == null && consolidatedJsonStr == null) {
                        MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.identify.eventMismatch", 0,
                                "issue:none");
                        continue;
                    } else if (originalJsonStr == null || consolidatedJsonStr == null) {
                        MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.identify.eventMismatch", 1,
                                "issue:onenull");
                        anyMismatch = true;
                        continue;
                    }

                    // Try comparing the strings first, as that will be more performant than converting to json for
                    // comparison
                    if (originalJsonStr.equals(consolidatedJsonStr)) {
                        continue;
                    }

                    // If strings don't match, convert to json so we can carefully check each key/value
                    JSONObject originalJson = JsonUtils.parseObject(originalJsonStr);
                    JSONObject consolidatedJson = JsonUtils.parseObject(consolidatedJsonStr);

                    JsonUtils.JSONEqualityResponse equalityResponse = JsonUtils.areJSONObjectsApproximatelyEqual(
                            originalJson,
                            consolidatedJson, RELATIVE_TOLERANCE, ABSOLUTE_TOLERANCE);
                    if (!equalityResponse.areEqual()) {
                        MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.identify.eventMismatch", 1,
                                "issue:jsonMismatch");
                        anyMismatch = true; // we have found at least one mismatch

                        if (Dynconfv2.instance().getBoolean(LOG_CONSOLIDATED_SHADOW_MISMATCH_DYNCONF_KEY, false)) {
                            // TODO: log these in a better way -- log basic details for now. We can't log the event
                            // because
                            // it may contain PII
                            LOGGER.info("Mismatched ident: app: {}. Reasons: {}", appId, equalityResponse.reasons());
                            if (Dynconfv2.instance().isAppInList(
                                    APPS_TO_LOG_EVENT_CONSOLIDATED_SHADOW_MISMATCH_DYNCONF_KEY,
                                    appId)) {
                                LOGGER.info("Ident: {}", originalJsonStr);
                                LOGGER.info("Consolidated Ident: {}", consolidatedJsonStr);
                            }
                        }
                    }
                }
            }
            return anyMismatch;
        } catch (Exception e) {
            LOGGER.error("Encountered error in compareConsolidatedIdentifyResponse for idents: " + e);
            MetricsWrapper.incrementCounter(
                    "taxonomy.consolidatedValidator.compareConsolidatedIdentifyResponse.errors", 1);
            return true;
        }
    }

    private static void compareWhiteListEntry(Map.Entry<Integer, TaxonomyWhitelistSettings> entry,
            Map<Integer, TaxonomyWhitelistSettings> taxonomyWhitelistSettings,
            String metricName) {
        TaxonomyWhitelistSettings settings = entry.getValue();
        TaxonomyWhitelistSettings consolidatedSettings = taxonomyWhitelistSettings
                .get(entry.getKey());
        int mismatch = settings.equals(consolidatedSettings) ? 0 : 1;
        if (mismatch == 1 && Dynconfv2.instance().getBoolean(LOG_CONSOLIDATED_SHADOW_MISMATCH_DYNCONF_KEY, false)) {
            LOGGER.info("(Shadow) Mismatched whitelist settings: " + settings + " vs " + consolidatedSettings);
        }
        MetricsWrapper.incrementCounter(metricName, mismatch);
    }

    private TSchemaValidation eventValidationHelperConsolidated(String type, List<String> events,
            TaxonomyServiceValidationMode validationMode,
            Map<Integer, Set<String>> clientKnownCheckSums, int appId) {
        long t0 = System.currentTimeMillis();
        boolean error = false;
        ValidationResponse validationResponse = null;
        try {
            LOGGER.debug("Received {} for {} events. (consolidated)", type, events.size());

            validationResponse = MetricsWrapper.instrumentMethod(() -> {
                BitSet unparseableEventIndexes = new BitSet(events.size());
                List<AmplitudeEventV13> parsedEvents = new ArrayList<>(events.size());
                for (int i = 0; i < events.size(); i++) {
                    try {
                        parsedEvents.add(AmplitudeEventV13.fromString(events.get(i)));
                    } catch (RuntimeException e) {
                        unparseableEventIndexes.set(i);
                    }
                }
                MetricsWrapper.incrementCounter("taxonomy.consolidated.unparseableEvents",
                        unparseableEventIndexes.cardinality());

                ValidationResponse response = consolidatedValidator.validateEvents(parsedEvents,
                        validationMode, clientKnownCheckSums);
                if (unparseableEventIndexes.isEmpty()) {
                    return response;
                }

                if (response.eventsOrPropertiesJsonStrings.size() != parsedEvents.size()) {
                    throw new IllegalStateException(String.format(
                            "Consolidated taxonomy returned %s event results for %s parsed events",
                            response.eventsOrPropertiesJsonStrings.size(), parsedEvents.size()));
                }

                List<String> alignedEvents = new ArrayList<>(events.size());
                int parsedEventIndex = 0;
                for (int i = 0; i < events.size(); i++) {
                    if (unparseableEventIndexes.get(i)) {
                        alignedEvents.add("null");
                    } else {
                        alignedEvents.add(response.eventsOrPropertiesJsonStrings.get(parsedEventIndex++));
                    }
                }
                response.eventsOrPropertiesJsonStrings = alignedEvents;
                LOGGER.warn("Dropped {} of {} events for app {} because they could not be parsed",
                        unparseableEventIndexes.cardinality(), events.size(), appId);
                return response;
            }, "taxonomy.consolidatedValidator.validateEvents", "shadow:off");
            return thriftResponseFromValidationResult(validationResponse);
        } catch (Exception e) {
            LOGGER.error("Encountered error in validateSchema (consolidated) for app " + appId + ", events: " + events,
                    e);
            error = true;
            throw e;
        } finally {
            long totalTimeMS = System.currentTimeMillis() - t0;
            if (!error) {
                Set<Integer> appsInRequest = new HashSet<>();
                if (validationResponse != null) {
                    appsInRequest = validationResponse.appIdToUserPropSchema.keySet();
                }

                LOGGER.debug("Finished (consolidated) {} for {} events in {} ms for apps {}", type, events.size(),
                        totalTimeMS,
                        appsInRequest);
            }
            MetricsWrapper.recordHistogram("taxonomy.validateSchema.ms", totalTimeMS, CONSOLIDATED_VERSION_TAG);
            MetricsWrapper.incrementCounter("taxonomy.validateSchema.errors", error ? 1 : 0, CONSOLIDATED_VERSION_TAG);

            // Bad data in dynconf will break the service here...
            if (totalTimeMS > Dynconfv2.instance().getLong("taxonomy.service.longValidationThresholdMS",
                    TimeUnit.SECONDS.toMillis(5))) {
                LOGGER.warn("LONG CONSOLIDATED VALIDATION {} {} events {} ms: {}", type, events.size(), totalTimeMS,
                        events);
            }
        }
    }

    private TSchemaValidation eventValidationHelper(String clientId, String type, List<String> events,
            boolean processForStargate, TaxonomyServiceValidationMode validationMode,
            Map<Integer, Set<String>> clientKnownCheckSums, int appId, int orgId) {
        long t0 = System.currentTimeMillis();
        boolean error = false;
        ValidationResponse validationResponse = null;
        try {
            LOGGER.debug("Received {} for {} events", type, events.size());

            // TODO(KURT): Cleaner error responses for client side errors, e.g. json parse error
            var jsonEvents = events.stream().map(event -> JsonUtils.fromJsonSimple(event, JSONObject.class))
                    .collect(Collectors.toList());
            Set<Integer> enabledApps = schemaValidator.getAppsWithTaxonomyValidationEnabled(jsonEvents);

            if (processForStargate) {
                // process stargate observe async and don't let it fail the validation call
                MetricsWrapper.instrumentMethod(() -> {
                    stargateObserve.processEventsAsync(clientId, jsonEvents, enabledApps);
                    return true;
                }, "taxonomy.stargate.processEvent");
            }

            validationResponse = MetricsWrapper.instrumentMethod(() -> {
                return schemaValidator.validateSchema(SchemaValidationType.fromString(type), jsonEvents,
                        validationMode, clientKnownCheckSums, enabledApps);
            }, "taxonomy.schemaValidator.validateSchema", LEGACY_VERSION_TAG);

            if (shouldShadowConsolidatedValidator(orgId)) {
                ValidationResponse finalValidationResponse = validationResponse;
                comparisonExecutor.execute(() -> {
                    try {
                        List<JSONObject> inputEventPropertiesList = jsonEvents.stream()
                                .map(event -> event.getJSONObject("event_properties"))
                                .collect(Collectors.toList());
                        List<JSONObject> inputUserPropertiesList = jsonEvents.stream().map(event -> event
                                .getJSONObject("user_properties")).collect(Collectors.toList());

                        ValidationResponse consolidatedResponse = MetricsWrapper.instrumentMethod(() -> {
                            List<AmplitudeEventV13> eventsV13 = events.stream()
                                    .map(event -> AmplitudeEventV13.fromString(event))
                                    .collect(Collectors.toList());
                            return consolidatedValidator.validateEvents(eventsV13,
                                    TaxonomyServiceValidationMode.DO_MUTATIONS_ONLY, clientKnownCheckSums);
                        }, "taxonomy.consolidatedValidator.validateEvents", "shadow:on");
                        compareConsolidatedResponse(appId, finalValidationResponse, consolidatedResponse,
                                inputEventPropertiesList, inputUserPropertiesList);
                    } catch (Exception e) {
                        LOGGER.error("Encountered error in consolidatedValidator for app: " + appId, e);
                        MetricsWrapper.incrementCounter("taxonomy.consolidatedValidator.errors", 1);
                    }
                });
            }
            return thriftResponseFromValidationResult(validationResponse);
        } catch (Exception e) {
            LOGGER.error("Encountered error in validateSchema for events: " + events, e);
            error = true;
            throw e;
        } finally {
            long totalTimeMS = System.currentTimeMillis() - t0;
            if (!error) {
                Set<Integer> appsInRequest = new HashSet<>();
                if (validationResponse != null) {
                    appsInRequest = validationResponse.appIdToUserPropSchema.keySet();
                }

                LOGGER.debug("Finished {} for {} events in {} ms for apps {}", type, events.size(), totalTimeMS,
                        appsInRequest);
            }
            MetricsWrapper.recordHistogram("taxonomy.validateSchema.ms", totalTimeMS, LEGACY_VERSION_TAG);
            MetricsWrapper.incrementCounter("taxonomy.validateSchema.errors", error ? 1 : 0, LEGACY_VERSION_TAG);

            // Bad data in dynconf will break the service here...
            if (totalTimeMS > Dynconfv2.instance().getLong("taxonomy.service.longValidationThresholdMS",
                    TimeUnit.SECONDS.toMillis(5))) {
                LOGGER.warn("LONG VALIDATION {} {} events {} ms: {}", type, events.size(), totalTimeMS, events);
            }
        }
    }

    @Override
    public String healthCheck(String clientId) throws TException {
        return "success " + clientId;
    }

    @Override
    public Map<String, String> getConfig() throws TException {
        Map<String, String> config = new HashMap<>();
        config.put("env", Config.ENV);
        return config;
    }

    @Override
    public TAppIsEnabledData isTaxonomyEnabled(int app) throws TException {
        TAppIsEnabledData tAppIsEnabledData = new TAppIsEnabledData();

        tAppIsEnabledData.setIsEnabled(schemaValidator.isTaxonomyEnabledApp(app));
        tAppIsEnabledData.setIsSchemaInitialized(schemaValidator.isSchemaInitializedInCoredata(app));

        // TODO (jeffrey): add stargate observe here

        return tAppIsEnabledData;
    }

    @Override
    public List<TEventValidationResult> validateBatchV3(int orgId, int appId, String clientId,
            List<List<String>> batch, boolean processForStargate,
            Map<Integer, Set<String>> clientKnownCheckSums) throws TException {
        // orgId is not in use. But it could be helpful in the future.
        int numBatches = 0;
        long numEvents = 0;
        boolean success = false;
        try {
            numBatches = batch.size();
            numEvents = batch.stream().mapToLong(List::size).sum();
            List<TEventValidationResult> result = MetricsWrapper.instrumentMethod(() -> {
                try {
                    TaxonomyServiceValidationMode validationMode = TaxonomyServiceValidationMode.DO_EVERYTHING;
                    List<TEventValidationResult> innerResult = new ArrayList<>();
                    for (List<String> events : batch) {
                        TSchemaValidation tSchemaValidation;
                        // If this app is enabled for consolidated rollout AND has the orion source of truth feature
                        // flag on, use the consolidated path
                        if (Dynconfv2.instance().isRolledOutForOrg(CONSOLIDATED_TAXONOMY_ROLLOUT_DYNCONF_KEY, orgId)
                                && isStargateOrionIngestionEnabledForOrg(orgId)) {
                            tSchemaValidation = eventValidationHelperConsolidated("event_validation", events,
                                    validationMode,
                                    clientKnownCheckSums, appId);
                        } else {
                            tSchemaValidation = eventValidationHelper(clientId, "event_validation", events,
                                    processForStargate, validationMode, clientKnownCheckSums, appId, orgId);
                        }

                        TEventValidationResult tEventValidationResult = new TEventValidationResult(
                                tSchemaValidation.sharedSchemaManagerId, tSchemaValidation.events,
                                tSchemaValidation.appIdToWhitelistSettings, tSchemaValidation.appIdToUserPropSchema);
                        tEventValidationResult
                                .setAppIdToUserPropSchemaChecksum(tSchemaValidation.appIdToUserPropSchemaChecksum);
                        tEventValidationResult
                                .setAppIdToDataplaneUserPropChecksum(
                                        tSchemaValidation.appIdToDataplaneUserPropChecksum);
                        tEventValidationResult
                                .setAppIdToDataplaneUserPropSchema(tSchemaValidation.appIdToDataplaneUserPropSchema);
                        innerResult.add(tEventValidationResult);
                    }
                    return innerResult;
                } catch (Exception e) {
                    LOGGER.error("Error while validating events", e);
                    MetricsWrapper.incrementCounter("taxonomy.validateSchema.errors", 1);
                    throw new RuntimeException("Error while validating events", e);
                }
            }, "taxonomy.validateBatchV3.ms", true, getLongStepThreshold(),
                    () -> String.format("[Taxonomy] app: %s, input: %s", appId,
                            batch));
            success = true;
            return result;
        } finally {
            recordValidateCounts(METHOD_TAG_VALIDATE_BATCH_V3, clientId, success, numBatches, numEvents);
        }
    }

    @Override
    public TIdentValidationResult validateUserProperties(String clientId, int app, List<String> serialUserProperties)
            throws TException {
        long numEvents = 0;
        boolean success = false;
        try {
            numEvents = serialUserProperties.size();
            TIdentValidationResult result = MetricsWrapper.instrumentMethod(() -> {
                try {
                    ValidationResponse validationResponse = schemaValidator.validateIdentificationUserProperties(app,
                            serialUserProperties.stream().map(JsonUtils::parseObject).collect(Collectors.toList()));

                    TSchemaValidation tSchemaValidation = thriftResponseFromValidationResult(validationResponse);

                    TIdentValidationResult tIdentValidationResult = new TIdentValidationResult();
                    tIdentValidationResult.setClientId(clientId);
                    // TODO(KURT): Fix all this plumbing so we better differentiate between events and user
                    // TODO(KURT): prop blobs OR just name things better
                    tIdentValidationResult.setUserProperties(tSchemaValidation.events);
                    tIdentValidationResult.setAppIdToUserPropSchema(tSchemaValidation.appIdToUserPropSchema);
                    tIdentValidationResult.setAppIdToWhitelistSettings(tSchemaValidation.appIdToWhitelistSettings);

                    return tIdentValidationResult;
                } catch (Exception e) {
                    LOGGER.error("Encountered error in validateSchema for idents: " + serialUserProperties, e);
                    throw e;
                }
            }, "taxonomy.validateUserProperties.ms", true, getLongStepThreshold(),
                    () -> String.format("[Taxonomy] app: %s, input: %s", app,
                            serialUserProperties));
            success = true;
            return result;
        } finally {
            recordValidateCounts(METHOD_TAG_VALIDATE_USER_PROPERTIES, clientId, success, 1, numEvents);
        }
    }

    private TBatchIdentValidationResult validateBatchUserPropertiesWithSchema(int appId, String clientId,
            List<List<String>> batch, PropertySchema<UserPropertyMetadata> userPropSchema) {
        TaxonomyWhitelistSettings whitelistSetting = schemaValidator.getWhitelistSettings(appId);
        List<List<String>> serialValidatedUserProperties = Lists.newArrayList();
        for (List<String> serialUserProperties : batch) {
            List<JSONObject> validatedUserProperties = schemaValidator.getValidatedUserProperties(appId,
                    serialUserProperties.stream().map(JsonUtils::parseObject).collect(Collectors.toList()),
                    whitelistSetting, userPropSchema);
            serialValidatedUserProperties.add(buildThriftEventsOrProperties(validatedUserProperties));
        }

        TBatchIdentValidationResult tBatchIdentValidationResult = new TBatchIdentValidationResult();
        tBatchIdentValidationResult.setClientId(clientId);
        // TODO(KURT): Fix all this plumbing so we better differentiate between events and user
        // TODO(KURT): prop blobs OR just name things better
        tBatchIdentValidationResult.setUserProperties(serialValidatedUserProperties);
        tBatchIdentValidationResult
                .setAppIdToWhitelistSettings(buildThriftWhitelistSettings(Map.of(appId, whitelistSetting)));

        return tBatchIdentValidationResult;
    }

    private TBatchIdentValidationResult validateBatchUserPropertiesOriginalWithShadow(int orgId, int appId,
            String clientId, List<List<String>> batch, Map<Integer, Set<String>> appIdToUserPropMetaChecksums) {
        UserPropertySchema userPropSchemaWithChecksum = schemaValidator
                .getUserPropSchemaWithChecksum(Set.of(appId)).get(appId);
        TBatchIdentValidationResult tBatchIdentValidationResult = validateBatchUserPropertiesWithSchema(appId,
                clientId, batch, userPropSchemaWithChecksum.getCustomUserPropertySchema());
        if (!(appIdToUserPropMetaChecksums.containsKey(appId)
                && appIdToUserPropMetaChecksums.get(appId).contains(userPropSchemaWithChecksum.getChecksum()))) {
            tBatchIdentValidationResult.setAppIdToUserPropSchema(buildThriftUserPropSchema(
                    Map.of(appId, userPropSchemaWithChecksum.getCustomPropSchemaWithGPStrippedKeys().values())));
        } else {
            tBatchIdentValidationResult.setAppIdToUserPropSchema(Map.of());
        }

        tBatchIdentValidationResult
                .setAppIdToUserPropSchemaChecksum(Map.of(appId, userPropSchemaWithChecksum.getChecksum()));

        if (shouldShadowConsolidatedValidator(orgId)) {
            comparisonExecutor.execute(() -> {
                IdentifyValidationResponse consolidatedResponse = MetricsWrapper.instrumentMethod(() -> {
                    // Validate each batch of user properties one at a time
                    try {
                        return consolidatedValidator.validateIdents(appId, batch, appIdToUserPropMetaChecksums,
                                TaxonomyServiceValidationMode.DO_MUTATIONS_ONLY);
                    } catch (Exception e) {
                        LOGGER.error(
                                "Encountered error in consolidated validateBatchUserPropertiesWithChecksum for idents: "
                                        + e);
                        MetricsWrapper.incrementCounter(
                                "taxonomy.consolidatedValidator.validateBatchUserPropertiesWithChecksum.errors", 1);
                        return null;
                    }
                }, "taxonomy.consolidatedValidator.validateEvents", "shadow:on");
                doesConsolidatedIdentifyResponseDiffer(appId, tBatchIdentValidationResult, consolidatedResponse);
            });
        }

        return tBatchIdentValidationResult;
    }

    private TBatchIdentValidationResult validateBatchUserPropertiesConsolidated(int appId,
            String clientId, List<List<String>> batch, Map<Integer, Set<String>> appIdToUserPropMetaChecksums) {
        IdentifyValidationResponse consolidatedResponse = MetricsWrapper.instrumentMethod(() -> {
            // Validate each batch of user properties one at a time
            try {
                return consolidatedValidator.validateIdents(appId, batch, appIdToUserPropMetaChecksums,
                        TaxonomyServiceValidationMode.DO_EVERYTHING);
            } catch (Exception e) {
                LOGGER.error(
                        "Encountered error in consolidated validateBatchUserPropertiesWithChecksum for idents: "
                                + e);
                MetricsWrapper.incrementCounter(
                        "taxonomy.consolidatedValidator.validateBatchUserPropertiesWithChecksum.errors", 1);
                return null;
            }
        }, "taxonomy.consolidatedValidator.validateEvents", "shadow:off");
        TBatchIdentValidationResult tBatchIdentValidationResult = new TBatchIdentValidationResult();
        tBatchIdentValidationResult.setClientId(clientId);
        // TODO: Fix all this plumbing so we better differentiate between events and user
        // TODO: prop blobs OR just name things better
        tBatchIdentValidationResult.setUserProperties(consolidatedResponse.jsonPropertiesBatches);
        tBatchIdentValidationResult
                .setAppIdToWhitelistSettings(
                        buildThriftWhitelistSettings(consolidatedResponse.taxonomyWhitelistSettings));
        tBatchIdentValidationResult.setAppIdToDataplaneUserPropSchema(
                buildThriftUserPropJsonSchema(consolidatedResponse.appIdToDataplaneUserPropSchema));
        tBatchIdentValidationResult
                .setAppIdToDataplaneUserPropChecksum(consolidatedResponse.appIdToUserPropSchemaChecksum);

        return tBatchIdentValidationResult;
    }

    @Override
    public TBatchIdentValidationResult validateBatchUserPropertiesWithChecksum(int orgId, int appId,
            String clientId, List<List<String>> batch, Map<Integer, Set<String>> appIdToUserPropMetaChecksums)
            throws TException {
        int numBatches = 0;
        long numEvents = 0;
        boolean success = false;
        try {
            numBatches = batch.size();
            numEvents = batch.stream().mapToLong(List::size).sum();
            TBatchIdentValidationResult result = MetricsWrapper.instrumentMethod(() -> {
                try {
                    // If this app is enabled for consolidated rollout AND has the orion source of truth feature flag
                    // on, use the consolidated path
                    if (Dynconfv2.instance().isRolledOutForOrg(CONSOLIDATED_TAXONOMY_ROLLOUT_DYNCONF_KEY, orgId)
                            && isStargateOrionIngestionEnabledForOrg(orgId)) {
                        return validateBatchUserPropertiesConsolidated(appId, clientId, batch,
                                appIdToUserPropMetaChecksums);
                    } else {
                        return validateBatchUserPropertiesOriginalWithShadow(orgId, appId, clientId, batch,
                                appIdToUserPropMetaChecksums);
                    }
                } catch (Exception e) {
                    LOGGER.error("Encountered error in validateBatchUserPropertiesWithChecksum for idents: " + batch,
                            e);
                    throw e;
                }
            }, "taxonomy.validateBatchUserPropertiesWithChecksum.ms", true, getLongStepThreshold(),
                    () -> String.format("[Taxonomy] app: %s, batch: %s", appId,
                            batch));
            success = true;
            return result;
        } finally {
            recordValidateCounts(METHOD_TAG_VALIDATE_BATCH_USER_PROPERTIES_WITH_CHECKSUM,
                    clientId, success, numBatches, numEvents);
        }
    }

    private static final String METHOD_TAG_VALIDATE_BATCH_V3 = "method:validateBatchV3";
    private static final String METHOD_TAG_VALIDATE_USER_PROPERTIES = "method:validateUserProperties";
    private static final String METHOD_TAG_VALIDATE_BATCH_USER_PROPERTIES_WITH_CHECKSUM = "method:validateBatchUserPropertiesWithChecksum";

    private static void recordValidateCounts(String methodTag, String clientId, boolean success,
            int numBatches, long numEvents) {
        String callerTag = "caller:" + clientId;
        String successTag = success ? "success:true" : "success:false";
        MetricsWrapper.incrementCounter("taxonomy.validate.requests", 1, methodTag, callerTag, successTag);
        MetricsWrapper.incrementCounter("taxonomy.validate.batches", numBatches, methodTag, callerTag, successTag);
        MetricsWrapper.incrementCounter("taxonomy.validate.events", numEvents, methodTag, callerTag, successTag);
    }

    private static long getLongStepThreshold() {
        return Dynconfv2.instance().getLong("taxonomy.service.longValidationThresholdMS", TimeUnit.SECONDS.toMillis(5));
    }
}