/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.atlas.repository.store.bootstrap;


import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import org.apache.atlas.AtlasErrorCode;
import org.apache.atlas.AtlasException;
import org.apache.atlas.RequestContext;
import org.apache.atlas.authorize.AtlasAuthorizerFactory;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.ha.HAConfiguration;
import org.apache.atlas.listener.ActiveStateChangeHandler;
import org.apache.atlas.model.TypeCategory;
import org.apache.atlas.model.patches.AtlasPatch.PatchStatus;
import org.apache.atlas.model.typedef.AtlasBaseTypeDef;
import org.apache.atlas.model.typedef.AtlasBusinessMetadataDef;
import org.apache.atlas.model.typedef.AtlasClassificationDef;
import org.apache.atlas.model.typedef.AtlasEntityDef;
import org.apache.atlas.model.typedef.AtlasEnumDef;
import org.apache.atlas.model.typedef.AtlasEnumDef.AtlasEnumElementDef;
import org.apache.atlas.model.typedef.AtlasRelationshipDef;
import org.apache.atlas.model.typedef.AtlasRelationshipDef.RelationshipCategory;
import org.apache.atlas.model.typedef.AtlasRelationshipEndDef;
import org.apache.atlas.model.typedef.AtlasStructDef;
import org.apache.atlas.model.typedef.AtlasStructDef.AtlasAttributeDef;
import org.apache.atlas.model.typedef.AtlasTypesDef;
import org.apache.atlas.repository.Constants;
import org.apache.atlas.repository.graph.GraphBackedSearchIndexer;
import org.apache.atlas.repository.graphdb.AtlasGraph;
import org.apache.atlas.repository.patches.AddMandatoryAttributesPatch;
import org.apache.atlas.repository.patches.SuperTypesUpdatePatch;
import org.apache.atlas.repository.patches.AtlasPatchManager;
import org.apache.atlas.repository.patches.AtlasPatchRegistry;
import org.apache.atlas.store.AtlasTypeDefStore;
import org.apache.atlas.type.AtlasEntityType;
import org.apache.atlas.type.AtlasStructType.AtlasAttribute;
import org.apache.atlas.type.AtlasType;
import org.apache.atlas.type.AtlasTypeRegistry;
import org.apache.atlas.type.AtlasTypeUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.inject.Inject;
import javax.xml.bind.annotation.XmlAccessType;
import javax.xml.bind.annotation.XmlAccessorType;
import javax.xml.bind.annotation.XmlRootElement;
import java.io.File;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.util.*;

import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.NONE;
import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.PUBLIC_ONLY;
import static org.apache.atlas.model.patches.AtlasPatch.PatchStatus.APPLIED;
import static org.apache.atlas.model.patches.AtlasPatch.PatchStatus.FAILED;
import static org.apache.atlas.model.patches.AtlasPatch.PatchStatus.SKIPPED;
import static org.apache.atlas.model.patches.AtlasPatch.PatchStatus.UNKNOWN;

/**
 * Class that handles initial loading of models and patches into typedef store
 */
@Service
@Order(2)
public class AtlasTypeDefStoreInitializer implements ActiveStateChangeHandler {
    public static final Logger LOG                    = LoggerFactory.getLogger(AtlasTypeDefStoreInitializer.class);
    public static final String PATCHES_FOLDER_NAME    = "patches";
    public static final String RELATIONSHIP_LABEL     = "relationshipLabel";
    public static final String RELATIONSHIP_CATEGORY  = "relationshipCategory";
    public static final String RELATIONSHIP_SWAP_ENDS = "swapEnds";
    public static final String TYPEDEF_PATCH_TYPE     = "TYPEDEF_PATCH";

    private final AtlasTypeDefStore typeDefStore;
    private final AtlasTypeRegistry typeRegistry;
    private final Configuration     conf;
    private final AtlasGraph        graph;
    private final AtlasPatchManager patchManager;

    @Inject
    public AtlasTypeDefStoreInitializer(AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry,
                                        AtlasGraph graph, Configuration conf, AtlasPatchManager patchManager) {
        this.typeDefStore  = typeDefStore;
        this.typeRegistry  = typeRegistry;
        this.conf          = conf;
        this.graph         = graph;
        this.patchManager  = patchManager;
    }

    @PostConstruct
    public void init() {
        LOG.info("==> AtlasTypeDefStoreInitializer.init()");

        if (!HAConfiguration.isHAEnabled(conf)) {
            startInternal();
        } else {
            LOG.info("AtlasTypeDefStoreInitializer.init(): deferring type loading until instance activation");
        }

        LOG.info("<== AtlasTypeDefStoreInitializer.init()");
    }

    /**
     * This method is looking for folders in alphabetical order in the models directory. It loads each of these folders and their associated patches in order.
     * It then loads any models in the top level folder and its patches.
     *
     * This allows models to be grouped into folders to help managability.
     *
     */
    private void loadBootstrapTypeDefs() {
        LOG.info("==> AtlasTypeDefStoreInitializer.loadBootstrapTypeDefs()");

        String atlasHomeDir  = System.getProperty("atlas.home");
        String modelsDirName = (StringUtils.isEmpty(atlasHomeDir) ? "." : atlasHomeDir) + File.separator + "models";

        if (modelsDirName == null || modelsDirName.length() == 0) {
            LOG.info("Types directory {} does not exist or not readable or has no typedef files", modelsDirName);
        } else {
            // look for folders we need to load models from
            File               topModeltypesDir  = new File(modelsDirName);
            File[]             modelsDirContents = topModeltypesDir.exists() ? topModeltypesDir.listFiles() : null;
            AtlasPatchRegistry patchRegistry     = new AtlasPatchRegistry(graph);

            if (modelsDirContents != null && modelsDirContents.length > 0) {
	            Arrays.sort(modelsDirContents);

	            for (File folder : modelsDirContents) {
	                    if (folder.isFile()) {
	                        // ignore files
	                        continue;
	                    } else if (!folder.getName().equals(PATCHES_FOLDER_NAME)){
	                        // load the models alphabetically in the subfolders apart from patches
	                        loadModelsInFolder(folder, patchRegistry);
	                    }
	            }
            }

            // load any files in the top models folder and any associated patches.
            loadModelsInFolder(topModeltypesDir, patchRegistry);
        }

        LOG.info("<== AtlasTypeDefStoreInitializer.loadBootstrapTypeDefs()");
    }

    /**
     * Load all the model files in the supplied folder followed by the contents of the patches folder.
     * @param typesDir
     */
    private void loadModelsInFolder(File typesDir, AtlasPatchRegistry patchRegistry) {
        LOG.info("==> AtlasTypeDefStoreInitializer({})", typesDir);

        String typesDirName = typesDir.getName();
        File[] typeDefFiles = typesDir.exists() ? typesDir.listFiles() : null;

        if (typeDefFiles == null || typeDefFiles.length == 0) {
            LOG.info("Types directory {} does not exist or not readable or has no typedef files", typesDirName );
        } else {
            // sort the files by filename
            Arrays.sort(typeDefFiles);

            for (File typeDefFile : typeDefFiles) {
                if (typeDefFile.isFile()) {
                    try {
                        String        jsonStr  = new String(Files.readAllBytes(typeDefFile.toPath()), StandardCharsets.UTF_8);
                        AtlasTypesDef typesDef = AtlasType.fromJson(jsonStr, AtlasTypesDef.class);

                        if (typesDef == null || typesDef.isEmpty()) {
                            LOG.info("No type in file {}", typeDefFile.getAbsolutePath());

                            continue;
                        }

                        AtlasTypesDef typesToCreate = getTypesToCreate(typesDef, typeRegistry);
                        AtlasTypesDef typesToUpdate = getTypesToUpdate(typesDef, typeRegistry, true);

                        if (!typesToCreate.isEmpty() || !typesToUpdate.isEmpty()) {
                            typeDefStore.createUpdateTypesDef(typesToCreate, typesToUpdate);

                            LOG.info("Created/Updated types defined in file {}", typeDefFile.getAbsolutePath());
                        } else {
                            LOG.info("No new type in file {}", typeDefFile.getAbsolutePath());
                        }

                    } catch (Throwable t) {
                        LOG.error("error while registering types in file {}", typeDefFile.getAbsolutePath(), t);
                    }
                }
            }

            applyTypePatches(typesDir.getPath(), patchRegistry);
        }
        LOG.info("<== AtlasTypeDefStoreInitializer({})", typesDir);
    }

    public static AtlasTypesDef getTypesToCreate(AtlasTypesDef typesDef, AtlasTypeRegistry typeRegistry) {
        AtlasTypesDef typesToCreate = new AtlasTypesDef();

        if (CollectionUtils.isNotEmpty(typesDef.getEnumDefs())) {
            for (AtlasEnumDef enumDef : typesDef.getEnumDefs()) {
                if (!typeRegistry.isRegisteredType(enumDef.getName())) {
                    typesToCreate.getEnumDefs().add(enumDef);
                }
            }
        }

        if (CollectionUtils.isNotEmpty(typesDef.getStructDefs())) {
            for (AtlasStructDef structDef : typesDef.getStructDefs()) {
                if (!typeRegistry.isRegisteredType(structDef.getName())) {
                    typesToCreate.getStructDefs().add(structDef);
                }
            }
        }

        if (CollectionUtils.isNotEmpty(typesDef.getClassificationDefs())) {
            for (AtlasClassificationDef classificationDef : typesDef.getClassificationDefs()) {
                if (!typeRegistry.isRegisteredType(classificationDef.getName())) {
                    typesToCreate.getClassificationDefs().add(classificationDef);
                }
            }
        }

        if (CollectionUtils.isNotEmpty(typesDef.getEntityDefs())) {
            for (AtlasEntityDef entityDef : typesDef.getEntityDefs()) {
                if (!typeRegistry.isRegisteredType(entityDef.getName())) {
                    typesToCreate.getEntityDefs().add(entityDef);
                }
            }
        }

        if (CollectionUtils.isNotEmpty(typesDef.getRelationshipDefs())) {
            for (AtlasRelationshipDef relationshipDef : typesDef.getRelationshipDefs()) {
                if (!typeRegistry.isRegisteredType(relationshipDef.getName())) {
                    typesToCreate.getRelationshipDefs().add(relationshipDef);
                }
            }
        }

        if (CollectionUtils.isNotEmpty(typesDef.getBusinessMetadataDefs())) {
            for (AtlasBusinessMetadataDef businessMetadataDef : typesDef.getBusinessMetadataDefs()) {
                if (!typeRegistry.isRegisteredType(businessMetadataDef.getName())) {
                    typesToCreate.getBusinessMetadataDefs().add(businessMetadataDef);
                }
            }
        }

        return typesToCreate;
    }

    public static AtlasTypesDef getTypesToUpdate(AtlasTypesDef typesDef, AtlasTypeRegistry typeRegistry, boolean checkTypeVersion) {
        AtlasTypesDef typesToUpdate = new AtlasTypesDef();

        if (CollectionUtils.isNotEmpty(typesDef.getStructDefs())) {
            for (AtlasStructDef newStructDef : typesDef.getStructDefs()) {
                AtlasStructDef  oldStructDef = typeRegistry.getStructDefByName(newStructDef.getName());

                if (oldStructDef == null) {
                    continue;
                }

                if (updateTypeAttributes(oldStructDef, newStructDef, checkTypeVersion)) {
                    typesToUpdate.getStructDefs().add(newStructDef);
                }
            }
        }

        if (CollectionUtils.isNotEmpty(typesDef.getClassificationDefs())) {
            for (AtlasClassificationDef newClassifDef : typesDef.getClassificationDefs()) {
                AtlasClassificationDef  oldClassifDef = typeRegistry.getClassificationDefByName(newClassifDef.getName());

                if (oldClassifDef == null) {
                    continue;
                }

                if (updateTypeAttributes(oldClassifDef, newClassifDef, checkTypeVersion)) {
                    typesToUpdate.getClassificationDefs().add(newClassifDef);
                }
            }
        }

        if (CollectionUtils.isNotEmpty(typesDef.getEntityDefs())) {
            for (AtlasEntityDef newEntityDef : typesDef.getEntityDefs()) {
                AtlasEntityDef  oldEntityDef = typeRegistry.getEntityDefByName(newEntityDef.getName());

                if (oldEntityDef == null) {
                    continue;
                }

                if (updateTypeAttributes(oldEntityDef, newEntityDef, checkTypeVersion)) {
                    typesToUpdate.getEntityDefs().add(newEntityDef);
                }
            }
        }

        if (CollectionUtils.isNotEmpty(typesDef.getEnumDefs())) {
            for (AtlasEnumDef newEnumDef : typesDef.getEnumDefs()) {
                AtlasEnumDef  oldEnumDef = typeRegistry.getEnumDefByName(newEnumDef.getName());

                if (oldEnumDef == null) {
                    continue;
                }

                if (isTypeUpdateApplicable(oldEnumDef, newEnumDef, checkTypeVersion)) {
                    if (CollectionUtils.isNotEmpty(oldEnumDef.getElementDefs())) {
                        for (AtlasEnumElementDef oldEnumElem : oldEnumDef.getElementDefs()) {
                            if (!newEnumDef.hasElement(oldEnumElem.getValue())) {
                                newEnumDef.addElement(oldEnumElem);
                            }
                        }
                    }

                    typesToUpdate.getEnumDefs().add(newEnumDef);
                }
            }
        }

        if (CollectionUtils.isNotEmpty(typesDef.getRelationshipDefs())) {
            for (AtlasRelationshipDef relationshipDef : typesDef.getRelationshipDefs()) {
                AtlasRelationshipDef  oldRelationshipDef = typeRegistry.getRelationshipDefByName(relationshipDef.getName());

                if (oldRelationshipDef == null) {
                    continue;
                }

                if (updateTypeAttributes(oldRelationshipDef, relationshipDef, checkTypeVersion)) {
                    typesToUpdate.getRelationshipDefs().add(relationshipDef);
                }
            }
        }

        if (CollectionUtils.isNotEmpty(typesDef.getBusinessMetadataDefs())) {
            for (AtlasBusinessMetadataDef businessMetadataDef : typesDef.getBusinessMetadataDefs()) {
                AtlasBusinessMetadataDef oldDef = typeRegistry.getBusinessMetadataDefByName(businessMetadataDef.getName());

                if (oldDef == null) {
                    continue;
                }

                if (updateTypeAttributes(oldDef, businessMetadataDef, checkTypeVersion)) {
                    typesToUpdate.getBusinessMetadataDefs().add(businessMetadataDef);
                }
            }
        }

        return typesToUpdate;
    }

    @Override
    public void instanceIsActive() {
        LOG.info("==> AtlasTypeDefStoreInitializer.instanceIsActive()");

        startInternal();

        LOG.info("<== AtlasTypeDefStoreInitializer.instanceIsActive()");
    }

    private void startInternal() {
        try {
            typeDefStore.init();
            loadBootstrapTypeDefs();
            typeDefStore.notifyLoadCompletion();
            try {
                AtlasAuthorizerFactory.getAtlasAuthorizer();
            } catch (Throwable t) {
                LOG.error("AtlasTypeDefStoreInitializer.instanceIsActive(): Unable to obtain AtlasAuthorizer", t);
            }
        } catch (AtlasBaseException e) {
            LOG.error("Failed to init after becoming active", e);
        } finally {
            RequestContext.clear();
        }
    }

    @Override
    public void instanceIsPassive() throws AtlasException {
        LOG.info("==> AtlasTypeDefStoreInitializer.instanceIsPassive()");

        LOG.info("<== AtlasTypeDefStoreInitializer.instanceIsPassive()");
    }

    @Override
    public int getHandlerOrder() {
        return HandlerOrder.TYPEDEF_STORE_INITIALIZER.getOrder();
    }

    private static boolean updateTypeAttributes(AtlasStructDef oldStructDef, AtlasStructDef newStructDef, boolean checkTypeVersion) {
        boolean ret = isTypeUpdateApplicable(oldStructDef, newStructDef, checkTypeVersion);

        if (ret) {
            // make sure that all attributes in oldDef are in newDef as well
            if (CollectionUtils.isNotEmpty(oldStructDef.getAttributeDefs())){
                for (AtlasAttributeDef oldAttrDef : oldStructDef.getAttributeDefs()) {
                    if (!newStructDef.hasAttribute(oldAttrDef.getName())) {
                        newStructDef.addAttribute(oldAttrDef);
                    }
                }
            }
        }

        return ret;
    }

    private static boolean isTypeUpdateApplicable(AtlasBaseTypeDef oldTypeDef, AtlasBaseTypeDef newTypeDef, boolean checkVersion) {
        boolean ret = true;

        if (checkVersion) {
            String oldTypeVersion = oldTypeDef.getTypeVersion();
            String newTypeVersion = newTypeDef.getTypeVersion();

            ret = ObjectUtils.compare(newTypeVersion, oldTypeVersion) > 0;
        }

        return ret;
    }

    private void applyTypePatches(String typesDirName, AtlasPatchRegistry patchRegistry) {
        String typePatchesDirName = typesDirName + File.separator + PATCHES_FOLDER_NAME;
        File   typePatchesDir     = new File(typePatchesDirName);
        File[] typePatchFiles     = typePatchesDir.exists() ? typePatchesDir.listFiles() : null;

        if (typePatchFiles == null || typePatchFiles.length == 0) {
            LOG.info("Type patches directory {} does not exist or not readable or has no patches", typePatchesDirName);
        } else {
            LOG.info("Type patches directory {} is being processed", typePatchesDirName);

            // sort the files by filename
            Arrays.sort(typePatchFiles);

            PatchHandler[] patchHandlers = new PatchHandler[] {
                    new UpdateEnumDefPatchHandler(typeDefStore, typeRegistry),
                    new AddAttributePatchHandler(typeDefStore, typeRegistry),
                    new UpdateAttributePatchHandler(typeDefStore, typeRegistry),
                    new RemoveLegacyRefAttributesPatchHandler(typeDefStore, typeRegistry),
                    new UpdateTypeDefOptionsPatchHandler(typeDefStore, typeRegistry),
                    new SetServiceTypePatchHandler(typeDefStore, typeRegistry),
                    new UpdateAttributeMetadataHandler(typeDefStore, typeRegistry),
                    new AddSuperTypePatchHandler(typeDefStore, typeRegistry),
                    new AddMandatoryAttributePatchHandler(typeDefStore, typeRegistry)
            };

            Map<String, PatchHandler> patchHandlerRegistry = new HashMap<>();

            for (PatchHandler patchHandler : patchHandlers) {
                for (String supportedAction : patchHandler.getSupportedActions()) {
                    patchHandlerRegistry.put(supportedAction, patchHandler);
                }
            }

            for (File typePatchFile : typePatchFiles) {
                if (typePatchFile.isFile()) {
                    String patchFile = typePatchFile.getAbsolutePath();

                    LOG.info("Applying patches in file {}", patchFile);

                    try {
                        String         jsonStr = new String(Files.readAllBytes(typePatchFile.toPath()), StandardCharsets.UTF_8);
                        TypeDefPatches patches = AtlasType.fromJson(jsonStr, TypeDefPatches.class);

                        if (patches == null || CollectionUtils.isEmpty(patches.getPatches())) {
                            LOG.info("No patches in file {}", patchFile);

                            continue;
                        }

                        int patchIndex = 0;
                        for (TypeDefPatch patch : patches.getPatches()) {
                            PatchHandler patchHandler = patchHandlerRegistry.get(patch.getAction());

                            if (patchHandler == null) {
                                LOG.error("Unknown patch action {} in file {}. Ignored", patch.getAction(), patchFile);
                                continue;
                            }

                            if (patchRegistry.isApplicable(patch.getId(), patchFile, patchIndex++)) {
                                PatchStatus status;

                                try {
                                    status = patchHandler.applyPatch(patch);
                                } catch (AtlasBaseException ex) {
                                    status = FAILED;

                                    LOG.error("Failed to apply {} (status: {}; action: {}) in file: {}. Ignored.",
                                               patch.getId(), status.toString(), patch.getAction(), patchFile);
                                }

                                patchRegistry.register(patch.id, patch.description, TYPEDEF_PATCH_TYPE, patch.action, status);
                                LOG.info("{} (status: {}; action: {}) in file: {}", patch.getId(), status.toString(), patch.getAction(), patchFile);
                            } else {
                                LOG.info("{} in file: {} already {}. Ignoring.", patch.getId(), patchFile, patchRegistry.getStatus(patch.getId()).toString());
                            }
                        }
                    } catch (Throwable t) {
                        LOG.error("Failed to apply patches in file {}. Ignored", patchFile, t);
                    }
                }
            }
        }
    }

    /**
     * typedef patch details
     */
    @JsonAutoDetect(getterVisibility = PUBLIC_ONLY, setterVisibility = PUBLIC_ONLY, fieldVisibility = NONE)
    @JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
    @JsonIgnoreProperties(ignoreUnknown = true)
    @XmlRootElement
    @XmlAccessorType(XmlAccessType.PROPERTY)
    static class TypeDefPatch {
        private String                  id;
        private String                  description;
        private String                  action;
        private String                  typeName;
        private String                  applyToVersion;
        private String                  updateToVersion;
        private Map<String, Object>     params;
        private List<AtlasAttributeDef> attributeDefs;
        private List<AtlasEnumElementDef> elementDefs;
        private Map<String, String>     typeDefOptions;
        private String                  serviceType;
        private String attributeName;
        private Set<String> superTypes;

        public String getId() {
            return id;
        }

        public void setId(String id) {
            this.id = id;
        }

        public String getDescription() {
            return description;
        }

        public void setDescription(String description) {
            this.description = description;
        }

        public String getAction() {
            return action;
        }

        public void setAction(String action) {
            this.action = action;
        }

        public String getTypeName() {
            return typeName;
        }

        public void setTypeName(String typeName) {
            this.typeName = typeName;
        }

        public String getApplyToVersion() {
            return applyToVersion;
        }

        public void setApplyToVersion(String applyToVersion) {
            this.applyToVersion = applyToVersion;
        }

        public String getUpdateToVersion() {
            return updateToVersion;
        }

        public void setUpdateToVersion(String updateToVersion) {
            this.updateToVersion = updateToVersion;
        }

        public Map<String, Object> getParams() {
            return params;
        }

        public void setParams(Map<String, Object> params) {
            this.params = params;
        }

        public List<AtlasAttributeDef> getAttributeDefs() {
            return attributeDefs;
        }

        public void setAttributeDefs(List<AtlasAttributeDef> attributeDefs) {
            this.attributeDefs = attributeDefs;
        }

        public List<AtlasEnumElementDef> getElementDefs() {
            return elementDefs;
        }

        public void setElementDefs(List<AtlasEnumElementDef> elementDefs) {
            this.elementDefs = elementDefs;
        }

        public Map<String, String> getTypeDefOptions() {
            return typeDefOptions;
        }

        public void setTypeDefOptions(Map<String, String> typeDefOptions) {
            this.typeDefOptions = typeDefOptions;
        }

        public String getServiceType() {
            return serviceType;
        }

        public void setServiceType(String serviceType) {
            this.serviceType = serviceType;
        }

        public String getAttributeName() { return attributeName; }

        public void setAttributeName(String attributeName) { this.attributeName = attributeName; }

        public Set<String> getSuperTypes() {
            return superTypes;
        }

        public void setSuperTypes(Set<String> superTypes) {
            this.superTypes = superTypes;
        }
    }

    /**
     * list of typedef patches
     */
    @JsonAutoDetect(getterVisibility = PUBLIC_ONLY, setterVisibility = PUBLIC_ONLY, fieldVisibility = NONE)
    @JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
    @JsonIgnoreProperties(ignoreUnknown = true)
    @XmlRootElement
    @XmlAccessorType(XmlAccessType.PROPERTY)
    static class TypeDefPatches {
        private List<TypeDefPatch> patches;

        public List<TypeDefPatch> getPatches() {
            return patches;
        }

        public void setPatches(List<TypeDefPatch> patches) {
            this.patches = patches;
        }
    }

    abstract class PatchHandler {
        protected final AtlasTypeDefStore typeDefStore;
        protected final AtlasTypeRegistry typeRegistry;
        protected final String[]          supportedActions;

        protected PatchHandler(AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry, String[] supportedActions) {
            this.typeDefStore     = typeDefStore;
            this.typeRegistry     = typeRegistry;
            this.supportedActions = supportedActions;
        }

        public String[] getSupportedActions() { return supportedActions; }

        public abstract PatchStatus applyPatch(TypeDefPatch patch) throws AtlasBaseException;

        protected boolean isPatchApplicable(TypeDefPatch patch, AtlasBaseTypeDef currentTypeDef) {
            String currentVersion = currentTypeDef.getTypeVersion();
            String applyToVersion = patch.getApplyToVersion();

            return currentVersion == null ||
                   currentVersion.equalsIgnoreCase(applyToVersion) ||
                   currentVersion.startsWith(applyToVersion + ".");
        }
    }

    class UpdateEnumDefPatchHandler extends PatchHandler {
        public UpdateEnumDefPatchHandler(AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry) {
            super(typeDefStore, typeRegistry, new String[]{"UPDATE_ENUMDEF"});
        }

        @Override
        public PatchStatus applyPatch(TypeDefPatch patch) throws AtlasBaseException {
            String typeName = patch.getTypeName();
            AtlasBaseTypeDef typeDef = typeRegistry.getTypeDefByName(typeName);
            PatchStatus ret;
            if (typeDef == null) {
                throw new AtlasBaseException(AtlasErrorCode.PATCH_FOR_UNKNOWN_TYPE, patch.getAction(), typeName);
            }
            if (isPatchApplicable(patch, typeDef)) {
                if (typeDef.getClass().equals(AtlasEnumDef.class)) {
                    AtlasEnumDef updatedDef = new AtlasEnumDef((AtlasEnumDef) typeDef);

                    for (AtlasEnumElementDef elementDef : patch.getElementDefs()) {
                        updatedDef.addElement(elementDef);
                    }
                    updatedDef.setTypeVersion(patch.getUpdateToVersion());
                    typeDefStore.updateEnumDefByName(typeName, updatedDef);
                    ret = APPLIED;
                } else {
                    throw new AtlasBaseException(AtlasErrorCode.PATCH_NOT_APPLICABLE_FOR_TYPE, patch.getAction(), typeDef.getClass().getSimpleName());
                }
            } else {
                LOG.info("patch skipped: typeName={}; applyToVersion={}; updateToVersion={}",
                        patch.getTypeName(), patch.getApplyToVersion(), patch.getUpdateToVersion());
                ret = SKIPPED;
            }
            return ret;
        }
    }

    class AddAttributePatchHandler extends PatchHandler {
        public AddAttributePatchHandler(AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry) {
            super(typeDefStore, typeRegistry, new String[] { "ADD_ATTRIBUTE" });
        }

            @Override
            public PatchStatus applyPatch(TypeDefPatch patch) throws AtlasBaseException {
            String           typeName       = patch.getTypeName();
            AtlasBaseTypeDef typeDef        = typeRegistry.getTypeDefByName(typeName);
            PatchStatus      ret;

            if (typeDef == null) {
                throw new AtlasBaseException(AtlasErrorCode.PATCH_FOR_UNKNOWN_TYPE, patch.getAction(), typeName);
            }

            if (isPatchApplicable(patch, typeDef)) {
                if (typeDef.getClass().equals(AtlasEntityDef.class)) {
                    AtlasEntityDef updatedDef = new AtlasEntityDef((AtlasEntityDef)typeDef);

                    for (AtlasAttributeDef attributeDef : patch.getAttributeDefs()) {
                        updatedDef.addAttribute(attributeDef);
                    }

                    updatedDef.setTypeVersion(patch.getUpdateToVersion());

                    typeDefStore.updateEntityDefByName(typeName, updatedDef);

                    ret = APPLIED;
                } else if (typeDef.getClass().equals(AtlasClassificationDef.class)) {
                    AtlasClassificationDef updatedDef = new AtlasClassificationDef((AtlasClassificationDef)typeDef);

                    for (AtlasAttributeDef attributeDef : patch.getAttributeDefs()) {
                        updatedDef.addAttribute(attributeDef);
                    }

                    updatedDef.setTypeVersion(patch.getUpdateToVersion());

                    typeDefStore.updateClassificationDefByName(typeName, updatedDef);

                    ret = APPLIED;
                } else if (typeDef.getClass().equals(AtlasStructDef.class)) {
                    AtlasStructDef updatedDef = new AtlasStructDef((AtlasStructDef)typeDef);

                    for (AtlasAttributeDef attributeDef : patch.getAttributeDefs()) {
                        updatedDef.addAttribute(attributeDef);
                    }

                    updatedDef.setTypeVersion(patch.getUpdateToVersion());

                    typeDefStore.updateStructDefByName(typeName, updatedDef);

                    ret = APPLIED;
                } else {
                    throw new AtlasBaseException(AtlasErrorCode.PATCH_NOT_APPLICABLE_FOR_TYPE, patch.getAction(), typeDef.getClass().getSimpleName());
                }
            } else {
                LOG.info("patch skipped: typeName={}; applyToVersion={}; updateToVersion={}",
                          patch.getTypeName(), patch.getApplyToVersion(), patch.getUpdateToVersion());

                ret = SKIPPED;

            }

            return ret;
        }
    }

    class AddMandatoryAttributePatchHandler extends PatchHandler {
        public AddMandatoryAttributePatchHandler(AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry) {
            super(typeDefStore, typeRegistry, new String[] { Constants.TYPEDEF_PATCH_ADD_MANDATORY_ATTRIBUTE });
        }

        @Override
        public PatchStatus applyPatch(TypeDefPatch patch) throws AtlasBaseException {
            String           typeName  = patch.getTypeName();
            AtlasBaseTypeDef typeDef   = typeRegistry.getTypeDefByName(typeName);
            PatchStatus      ret;

            if (typeDef == null) {
                throw new AtlasBaseException(AtlasErrorCode.PATCH_FOR_UNKNOWN_TYPE, patch.getAction(), typeName);
            }

            if (isPatchApplicable(patch, typeDef)) {
                List<AtlasAttributeDef> attributesToAdd = getAttributesToAdd(patch, (AtlasStructDef) typeDef);

                if (CollectionUtils.isEmpty(attributesToAdd)) {
                    LOG.info("patch skipped: typeName={}; mandatory attributes are not valid in patch {}",patch.getTypeName(), patch.getId());

                    ret = SKIPPED;
                } else {
                    try {
                        RequestContext.get().setInTypePatching(true);

                        RequestContext.get().setCurrentTypePatchAction(Constants.TYPEDEF_PATCH_ADD_MANDATORY_ATTRIBUTE);

                        if (typeDef.getClass().equals(AtlasEntityDef.class)) {
                            AtlasEntityDef updatedDef = new AtlasEntityDef((AtlasEntityDef) typeDef);

                            updateTypeDefWithPatch(patch, updatedDef, attributesToAdd);

                            typeDefStore.updateEntityDefByName(typeName, updatedDef);
                        } else if (typeDef.getClass().equals(AtlasClassificationDef.class)) {
                            AtlasClassificationDef updatedDef = new AtlasClassificationDef((AtlasClassificationDef) typeDef);

                            updateTypeDefWithPatch(patch, updatedDef, attributesToAdd);

                            typeDefStore.updateClassificationDefByName(typeName, updatedDef);
                        } else if (typeDef.getClass().equals(AtlasStructDef.class)) {
                            AtlasStructDef updatedDef = new AtlasStructDef((AtlasStructDef) typeDef);

                            updateTypeDefWithPatch(patch, updatedDef, attributesToAdd);

                            typeDefStore.updateStructDefByName(typeName, updatedDef);
                        } else {
                            throw new AtlasBaseException(AtlasErrorCode.PATCH_NOT_APPLICABLE_FOR_TYPE, patch.getAction(), typeDef.getClass().getSimpleName());
                        }

                        LOG.info("adding a Java patch to update entities of {} with new mandatory attributes", typeName);

                        // Java patch handler to add mandatory attributes
                        patchManager.addPatchHandler(new AddMandatoryAttributesPatch(patchManager.getContext(), patch.getId(), typeName, attributesToAdd));

                        ret = APPLIED;
                    } finally {
                        RequestContext.get().setInTypePatching(false);

                        RequestContext.clear();
                    }
                }
            } else {
                LOG.info("patch skipped: typeName={}; applyToVersion={}; updateToVersion={}",
                        patch.getTypeName(), patch.getApplyToVersion(), patch.getUpdateToVersion());

                ret = SKIPPED;

            }

            return ret;
        }

        // Validate mandatory attribute with non-empty default value if PRIMITIVE, not unique and doesn't exists
        private List<AtlasAttributeDef> getAttributesToAdd(TypeDefPatch patch, AtlasStructDef updatedDef) throws AtlasBaseException {
            List<AtlasAttributeDef> ret = new ArrayList<>();

            for (AtlasAttributeDef attributeDef : patch.getAttributeDefs()) {
                TypeCategory attributeType = typeRegistry.getType(attributeDef.getTypeName()).getTypeCategory();

                if (updatedDef.hasAttribute(attributeDef.getName())) {
                    LOG.warn("AddMandatoryAttributePatchHandler(id={}, typeName={}, attribute={}): already exists in type {}. Ignoring attribute", patch.getId(), patch.getTypeName(), attributeDef.getName(), updatedDef.getName());
                } else if (attributeDef.getIsOptional()) {
                    LOG.warn("AddMandatoryAttributePatchHandler(id={}, typeName={}, attribute={}): is not mandatory attribute. Ignoring attribute", patch.getId(), patch.getTypeName(), attributeDef.getName());
                } else if (StringUtils.isEmpty(attributeDef.getDefaultValue())) {
                    LOG.warn("AddMandatoryAttributePatchHandler(id={}, typeName={}, attribute={}): default value is missing. Ignoring attribute", patch.getId(), patch.getTypeName(), attributeDef.getName());
                } else if (!TypeCategory.PRIMITIVE.equals(attributeType)) {
                    LOG.warn("AddMandatoryAttributePatchHandler(id={}, typeName={}, attribute={}): type {} is not primitive. Ignoring attribute", patch.getId(), patch.getTypeName(), attributeDef.getName(), attributeDef.getTypeName());
                } else if (attributeDef.getIsUnique()) {
                    LOG.warn("AddMandatoryAttributePatchHandler(id={}, typeName={}, attribute={}): is not unique. Ignoring attribute", patch.getId(), patch.getTypeName(), attributeDef.getName());
                } else {
                    ret.add(attributeDef);
                }
            }

            return ret;
        }

        private void updateTypeDefWithPatch(TypeDefPatch patch, AtlasStructDef updatedDef, List<AtlasAttributeDef> attributesToAdd) {
            for (AtlasAttributeDef attributeDef : attributesToAdd) {
                updatedDef.addAttribute(attributeDef);
            }

            updatedDef.setTypeVersion(patch.getUpdateToVersion());
        }
    }

    class UpdateAttributePatchHandler extends PatchHandler {
        public UpdateAttributePatchHandler(AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry) {
            super(typeDefStore, typeRegistry, new String[] { "UPDATE_ATTRIBUTE" });
        }

        @Override
        public PatchStatus applyPatch(TypeDefPatch patch) throws AtlasBaseException {
            String           typeName       = patch.getTypeName();
            AtlasBaseTypeDef typeDef        = typeRegistry.getTypeDefByName(typeName);
            PatchStatus      ret;

            if (typeDef == null) {
                throw new AtlasBaseException(AtlasErrorCode.PATCH_FOR_UNKNOWN_TYPE, patch.getAction(), typeName);
            }

            if (isPatchApplicable(patch, typeDef)) {
                if (typeDef.getClass().equals(AtlasEntityDef.class)) {
                    AtlasEntityDef updatedDef = new AtlasEntityDef((AtlasEntityDef)typeDef);

                    addOrUpdateAttributes(updatedDef, patch.getAttributeDefs());

                    updatedDef.setTypeVersion(patch.getUpdateToVersion());

                    typeDefStore.updateEntityDefByName(typeName, updatedDef);

                    ret = APPLIED;
                } else if (typeDef.getClass().equals(AtlasClassificationDef.class)) {
                    AtlasClassificationDef updatedDef = new AtlasClassificationDef((AtlasClassificationDef)typeDef);

                    addOrUpdateAttributes(updatedDef, patch.getAttributeDefs());

                    updatedDef.setTypeVersion(patch.getUpdateToVersion());

                    typeDefStore.updateClassificationDefByName(typeName, updatedDef);

                    ret = APPLIED;
                } else if (typeDef.getClass().equals(AtlasStructDef.class)) {
                    AtlasStructDef updatedDef = new AtlasStructDef((AtlasStructDef)typeDef);

                    addOrUpdateAttributes(updatedDef, patch.getAttributeDefs());

                    updatedDef.setTypeVersion(patch.getUpdateToVersion());

                    typeDefStore.updateStructDefByName(typeName, updatedDef);

                    ret = APPLIED;
                } else {
                    throw new AtlasBaseException(AtlasErrorCode.PATCH_NOT_APPLICABLE_FOR_TYPE, patch.getAction(), typeDef.getClass().getSimpleName());
                }
            } else {
                LOG.info("patch skipped: typeName={}; applyToVersion={}; updateToVersion={}",
                          patch.getTypeName(), patch.getApplyToVersion(), patch.getUpdateToVersion());

                ret = SKIPPED;
            }

            return ret;
        }

        private void addOrUpdateAttributes(AtlasStructDef structDef, List<AtlasAttributeDef> attributesToUpdate) {
            for (AtlasAttributeDef attributeToUpdate : attributesToUpdate) {
                String attrName = attributeToUpdate.getName();

                if (structDef.hasAttribute(attrName)) {
                    structDef.removeAttribute(attrName);
                }

                structDef.addAttribute(attributeToUpdate);
            }
        }
    }

    class RemoveLegacyRefAttributesPatchHandler extends PatchHandler {
        public RemoveLegacyRefAttributesPatchHandler(AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry) {
            super(typeDefStore, typeRegistry, new String[] { "REMOVE_LEGACY_REF_ATTRIBUTES" });
        }

        @Override
        public PatchStatus applyPatch(TypeDefPatch patch) throws AtlasBaseException {
            String           typeName       = patch.getTypeName();
            AtlasBaseTypeDef typeDef        = typeRegistry.getTypeDefByName(typeName);
            PatchStatus      ret            = UNKNOWN;

            if (typeDef == null) {
                throw new AtlasBaseException(AtlasErrorCode.PATCH_FOR_UNKNOWN_TYPE, patch.getAction(), typeName);
            }

            if (isPatchApplicable(patch, typeDef)) {
                if (typeDef.getClass().equals(AtlasRelationshipDef.class)) {
                    AtlasRelationshipDef    relationshipDef = (AtlasRelationshipDef) typeDef;
                    AtlasRelationshipEndDef end1Def         = relationshipDef.getEndDef1();
                    AtlasRelationshipEndDef end2Def         = relationshipDef.getEndDef2();
                    AtlasEntityType         end1Type        = typeRegistry.getEntityTypeByName(end1Def.getType());
                    AtlasEntityType         end2Type        = typeRegistry.getEntityTypeByName(end2Def.getType());

                    String               newRelationshipLabel    = null;
                    RelationshipCategory newRelationshipCategory = null;
                    boolean              swapEnds                = false;

                    if (patch.getParams() != null) {
                        Object relLabel    = patch.getParams().get(RELATIONSHIP_LABEL);
                        Object relCategory = patch.getParams().get(RELATIONSHIP_CATEGORY);
                        Object relSwapEnds = patch.getParams().get(RELATIONSHIP_SWAP_ENDS);

                        if (relLabel != null) {
                            newRelationshipLabel = relLabel.toString();
                        }

                        if (relCategory != null) {
                            newRelationshipCategory = RelationshipCategory.valueOf(relCategory.toString());
                        }

                        if (relSwapEnds != null) {
                            swapEnds = Boolean.valueOf(relSwapEnds.toString());
                        }
                    }

                    if (StringUtils.isEmpty(newRelationshipLabel)) {
                        if (end1Def.getIsLegacyAttribute()) {
                            if (!end2Def.getIsLegacyAttribute()) {
                                AtlasAttribute legacyAttribute = end1Type.getAttribute(end1Def.getName());

                                newRelationshipLabel = "__" + legacyAttribute.getQualifiedName();
                            } else { // if both ends are legacy attributes, RELATIONSHIP_LABEL should be specified in the patch
                                throw new AtlasBaseException(AtlasErrorCode.PATCH_MISSING_RELATIONSHIP_LABEL, patch.getAction(), typeName);
                            }
                        } else if (end2Def.getIsLegacyAttribute()) {
                            AtlasAttribute legacyAttribute = end2Type.getAttribute(end2Def.getName());

                            newRelationshipLabel = "__" + legacyAttribute.getQualifiedName();
                        } else {
                            newRelationshipLabel = relationshipDef.getRelationshipLabel();
                        }
                    }

                    AtlasRelationshipDef updatedDef = new AtlasRelationshipDef(relationshipDef);

                    if (swapEnds) {
                        AtlasRelationshipEndDef tmp = updatedDef.getEndDef1();

                        updatedDef.setEndDef1(updatedDef.getEndDef2());
                        updatedDef.setEndDef2(tmp);
                    }

                    end1Def  = updatedDef.getEndDef1();
                    end2Def  = updatedDef.getEndDef2();
                    end1Type = typeRegistry.getEntityTypeByName(end1Def.getType());
                    end2Type = typeRegistry.getEntityTypeByName(end2Def.getType());

                    updatedDef.setRelationshipLabel(newRelationshipLabel);

                    if (newRelationshipCategory != null) {
                        updatedDef.setRelationshipCategory(newRelationshipCategory);
                    }

                    updatedDef.setTypeVersion(patch.getUpdateToVersion());

                    AtlasEntityDef updatedEntityDef1 = new AtlasEntityDef(end1Type.getEntityDef());
                    AtlasEntityDef updatedEntityDef2 = new AtlasEntityDef(end2Type.getEntityDef());

                    updatedEntityDef1.removeAttribute(end1Def.getName());
                    updatedEntityDef2.removeAttribute(end2Def.getName());

                    AtlasTypesDef typesDef = new AtlasTypesDef();

                    typesDef.setEntityDefs(Arrays.asList(updatedEntityDef1, updatedEntityDef2));
                    typesDef.setRelationshipDefs(Collections.singletonList(updatedDef));

                    try {
                        RequestContext.get().setInTypePatching(true); // to allow removal of attributes

                        typeDefStore.updateTypesDef(typesDef);

                        ret = APPLIED;
                    } finally {
                        RequestContext.get().setInTypePatching(false);
                        RequestContext.clear();
                    }
                }
            } else {
                LOG.info("patch skipped: typeName={}; applyToVersion={}; updateToVersion={}",
                         patch.getTypeName(), patch.getApplyToVersion(), patch.getUpdateToVersion());

                ret = SKIPPED;
            }

            return ret;
        }
    }

    class UpdateTypeDefOptionsPatchHandler extends PatchHandler {
        public UpdateTypeDefOptionsPatchHandler(AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry) {
            super(typeDefStore, typeRegistry, new String[] { "UPDATE_TYPEDEF_OPTIONS" });
        }

        @Override
        public PatchStatus applyPatch(TypeDefPatch patch) throws AtlasBaseException {
            String           typeName = patch.getTypeName();
            AtlasBaseTypeDef typeDef  = typeRegistry.getTypeDefByName(typeName);
            PatchStatus      ret;

            if (typeDef == null) {
                throw new AtlasBaseException(AtlasErrorCode.PATCH_FOR_UNKNOWN_TYPE, patch.getAction(), typeName);
            }

            if (MapUtils.isEmpty(patch.getTypeDefOptions())) {
                throw new AtlasBaseException(AtlasErrorCode.PATCH_INVALID_DATA, patch.getAction(), typeName);
            }

            if (isPatchApplicable(patch, typeDef)) {
                if (typeDef.getOptions() == null) {
                    typeDef.setOptions(patch.getTypeDefOptions());
                } else {
                    typeDef.getOptions().putAll(patch.getTypeDefOptions());
                }
                typeDef.setTypeVersion(patch.getUpdateToVersion());

                typeDefStore.updateTypesDef(AtlasTypeUtil.getTypesDef(typeDef));

                ret = APPLIED;
            } else {
                LOG.info("patch skipped: typeName={}; applyToVersion={}; updateToVersion={}",
                         patch.getTypeName(), patch.getApplyToVersion(), patch.getUpdateToVersion());

                ret = SKIPPED;
            }

            return ret;
        }
    }

    class SetServiceTypePatchHandler extends PatchHandler {
        public SetServiceTypePatchHandler(AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry) {
            super(typeDefStore, typeRegistry, new String[] { "SET_SERVICE_TYPE" });
        }

        @Override
        public PatchStatus applyPatch(TypeDefPatch patch) throws AtlasBaseException {
            String           typeName       = patch.getTypeName();
            AtlasBaseTypeDef typeDef        = typeRegistry.getTypeDefByName(typeName);
            PatchStatus      ret;

            if (typeDef == null) {
                throw new AtlasBaseException(AtlasErrorCode.PATCH_FOR_UNKNOWN_TYPE, patch.getAction(), typeName);
            }

            if (isPatchApplicable(patch, typeDef)) {
                typeDef.setServiceType(patch.getServiceType());
                typeDef.setTypeVersion(patch.getUpdateToVersion());

                typeDefStore.updateTypesDef(AtlasTypeUtil.getTypesDef(typeDef));

                ret = APPLIED;
            } else {
                LOG.info("patch skipped: typeName={}; applyToVersion={}; updateToVersion={}", patch.getTypeName(),
                          patch.getApplyToVersion(), patch.getUpdateToVersion());

                ret = SKIPPED;
            }

            return ret;
        }
    }

    class UpdateAttributeMetadataHandler extends PatchHandler {
        public UpdateAttributeMetadataHandler(AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry) {
            super(typeDefStore, typeRegistry, new String[] { "UPDATE_ATTRIBUTE_METADATA" });
        }

        @Override
        public PatchStatus applyPatch(TypeDefPatch patch) throws AtlasBaseException {
            String           typeName       = patch.getTypeName();
            AtlasBaseTypeDef typeDef        = typeRegistry.getTypeDefByName(typeName);
            PatchStatus      ret;

            if (typeDef == null) {
                throw new AtlasBaseException(AtlasErrorCode.PATCH_FOR_UNKNOWN_TYPE, patch.getAction(), typeName);
            }

            if (isPatchApplicable(patch, typeDef)) {
                String attributeNameFromPatch = patch.getAttributeName();
                if (typeDef.getClass().equals(AtlasEntityDef.class)) {
                    AtlasEntityDef entityDef = new AtlasEntityDef((AtlasEntityDef)typeDef);

                    updateAttributeMetadata(patch, entityDef.getAttributeDefs());

                    entityDef.setTypeVersion(patch.getUpdateToVersion());

                    typeDefStore.updateEntityDefByName(typeName, entityDef);

                    ret = APPLIED;
                } else if (typeDef.getClass().equals(AtlasStructDef.class)) {
                    AtlasStructDef updatedDef = new AtlasStructDef((AtlasStructDef)typeDef);

                    updateAttributeMetadata(patch, updatedDef.getAttributeDefs());

                    updatedDef.setTypeVersion(patch.getUpdateToVersion());

                    typeDefStore.updateStructDefByName(typeName, updatedDef);

                    ret = APPLIED;
                } else {
                    throw new AtlasBaseException(AtlasErrorCode.PATCH_NOT_APPLICABLE_FOR_TYPE, patch.getAction(), typeDef.getClass().getSimpleName());
                }
            } else {
                LOG.info("patch skipped: typeName={}; applyToVersion={}; updateToVersion={}",
                        patch.getTypeName(), patch.getApplyToVersion(), patch.getUpdateToVersion());
                ret = SKIPPED;
            }

            return ret;
        }

        private void updateAttributeMetadata(TypeDefPatch patch, List<AtlasAttributeDef> attributeDefsFromEntity) {
            for(AtlasAttributeDef attributeDef: attributeDefsFromEntity) {
                if(attributeDef.getName().equalsIgnoreCase(patch.getAttributeName())) {
                    updateAttribute(attributeDef, patch.getParams());
                }
            }
        }

        private void updateAttribute(AtlasAttributeDef atlasAttributeDef, Map<String, Object> params) {
            if(!params.isEmpty()) {
                for(Map.Entry<String, Object> entry: params.entrySet()) {
                    try {
                        if (AtlasAttributeDef.SEARCH_WEIGHT_ATTR_NAME.equalsIgnoreCase(entry.getKey())) {
                            Number number = (Number) entry.getValue();
                            int searchWeight = number.intValue();
                            if(!GraphBackedSearchIndexer.isValidSearchWeight(number.intValue())) {
                                String msg = String.format("Invalid search weight '%d' was provided for property %s.", searchWeight, atlasAttributeDef.getName());
                                LOG.error(msg);
                                throw new RuntimeException(msg);
                            }
                            atlasAttributeDef.setSearchWeight(searchWeight);
                            LOG.info("Updating Model attribute {}'s property{} to {}.", atlasAttributeDef.getName(), entry.getKey(), entry.getValue());
                        } else if (AtlasAttributeDef.INDEX_TYPE_ATTR_NAME.equalsIgnoreCase(entry.getKey())) {
                            String indexTypeString = (String) entry.getValue();
                            if(!StringUtils.isEmpty(indexTypeString)) {
                                try {
                                    AtlasAttributeDef.IndexType indexType = AtlasAttributeDef.IndexType.valueOf(indexTypeString);
                                    atlasAttributeDef.setIndexType(indexType);
                                } catch (IllegalArgumentException e) {
                                    String msg = String.format("Value %s provided for the attribute %s is not valid.", indexTypeString, AtlasAttributeDef.INDEX_TYPE_ATTR_NAME);
                                    LOG.error(msg);
                                    throw new RuntimeException(msg);
                                }
                            }
                        } else {
                            //sanity exception
                            //more attributes can be added as needed.
                            String msg = String.format("Received unknown property{} for attribute {}'s ", entry.getKey(), atlasAttributeDef.getName());
                            LOG.error(msg);
                            throw new RuntimeException(msg);
                        }
                    } catch (Exception e) {
                        String msg = String.format("Error encountered in updating Model attribute %s's property '%s' to %s.", atlasAttributeDef.getName(), entry.getKey(), entry.getValue().toString());
                        LOG.info(msg, e);
                        throw new RuntimeException(msg, e);
                    }
                }
            }
        }
    }

    class AddSuperTypePatchHandler extends PatchHandler {

        public AddSuperTypePatchHandler(AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry) {
            super(typeDefStore, typeRegistry, new String[] {"ADD_SUPER_TYPES"});
        }

        @Override
        public PatchStatus applyPatch(TypeDefPatch patch) throws AtlasBaseException {
            String           typeName = patch.getTypeName();
            AtlasBaseTypeDef typeDef  = typeRegistry.getTypeDefByName(typeName);
            PatchStatus      ret;

            if (typeDef == null) {
                throw new AtlasBaseException(AtlasErrorCode.PATCH_FOR_UNKNOWN_TYPE, patch.getAction(), typeName);
            }

            Set<String> superTypesToBeAdded = patch.getSuperTypes();

            if (CollectionUtils.isNotEmpty(superTypesToBeAdded) && isPatchApplicable(patch, typeDef)) {
                if (typeDef.getClass().equals(AtlasEntityDef.class)) {
                    AtlasEntityDef updatedDef = new AtlasEntityDef((AtlasEntityDef)typeDef);

                    for (String superType : superTypesToBeAdded) {
                        updatedDef.addSuperType(superType);
                    }

                    updatedDef.setTypeVersion(patch.getUpdateToVersion());

                    typeDefStore.updateEntityDefByName(typeName, updatedDef);

                    LOG.info("Update entities of {} with new supertypes", typeName);

                    // add to java patch handlers to update entity supertypes
                    patchManager.addPatchHandler(new SuperTypesUpdatePatch(patchManager.getContext(), patch.getId(), typeName));

                    ret = APPLIED;
                } else {
                    throw new AtlasBaseException(AtlasErrorCode.PATCH_NOT_APPLICABLE_FOR_TYPE, patch.getAction(), typeDef.getClass().getSimpleName());
                }
            } else {
                if (CollectionUtils.isEmpty(superTypesToBeAdded)) {
                    LOG.info("patch skipped: No superTypes provided to add for typeName={}", patch.getTypeName());
                } else {
                    LOG.info("patch skipped: typeName={}; applyToVersion={}; updateToVersion={}",
                            patch.getTypeName(), patch.getApplyToVersion(), patch.getUpdateToVersion());
                }

                ret = SKIPPED;
            }

            return ret;
        }
    }
}