AMBARI-13398. Implement REST resource for storing Kerberos descriptors (Laszlo Puskas via smohanty)

ambari-server/pom.xml

@@ -1884,7 +1884,7 @@
     <dependency>
       <groupId>org.easymock</groupId>
       <artifactId>easymock</artifactId>
-      <version>3.1</version>
+      <version>3.3</version>
       <scope>test</scope>
     </dependency>
     <dependency>

ambari-server/src/main/java/org/apache/ambari/server/api/resources/ResourceInstanceFactoryImpl.java

@@ -6,9 +6,9 @@
  * 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
- *
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
  * 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.
@@ -388,6 +388,10 @@ public class ResourceInstanceFactoryImpl implements ResourceInstanceFactory {
         resourceDefinition = new HostKerberosIdentityResourceDefinition();
         break;
 
+      case KerberosDescriptor:
+        resourceDefinition = new SimpleResourceDefinition(Resource.Type.KerberosDescriptor, "kerberos_descriptor", "kerberos_descriptors");
+        break;
+
       case Credential:
         resourceDefinition = new CredentialResourceDefinition();
         break;

ambari-server/src/main/java/org/apache/ambari/server/api/services/KerberosDescriptorService.java

@@ -0,0 +1,90 @@
+package org.apache.ambari.server.api.services;
+
+import org.apache.ambari.server.api.resources.ResourceInstance;
+import org.apache.ambari.server.controller.spi.Resource;
+
+import javax.ws.rs.DELETE;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.HttpHeaders;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.UriInfo;
+import java.util.Collections;
+
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.
+ */
+
+@Path("/kerberos_descriptors/")
+public class KerberosDescriptorService extends BaseService {
+
+  /**
+   * Handles: GET  /kerberos_descriptors
+   * Get all kerberos descriptors.
+   *
+   * @param headers http headers
+   * @param ui      uri info
+   * @return a collection of kerberos descriptors
+   */
+  @GET
+  @Produces("text/plain")
+  public Response getKerberosDescriptors(String body, @Context HttpHeaders headers, @Context UriInfo ui) {
+    return handleRequest(headers, body, ui, Request.Type.GET, createKerberosDescriptorResource(null));
+  }
+
+  @GET
+  @Path("{kerberosDescriptorName}")
+  @Produces("text/plain")
+  public Response getKerberosDescriptor(String body, @Context HttpHeaders headers, @Context UriInfo ui,
+                                        @PathParam("kerberosDescriptorName") String kerberosDescriptorName) {
+    return handleRequest(headers, body, ui, Request.Type.GET, createKerberosDescriptorResource(kerberosDescriptorName));
+  }
+
+  @POST
+  @Path("{kerberosDescriptorName}")
+  @Produces("text/plain")
+  public Response createKerberosDescriptor(String body, @Context HttpHeaders headers, @Context UriInfo ui,
+                                           @PathParam("kerberosDescriptorName") String kerberosDescriptorName) {
+    return handleRequest(headers, body, ui, Request.Type.POST, createKerberosDescriptorResource(kerberosDescriptorName));
+  }
+
+  /**
+   * Handles: DELETE /kerberos_descriptors/{kerberosDescriptorName}
+   * Delete a specific kerberos descriptor.
+   *
+   * @param headers                http headers
+   * @param ui                     uri info
+   * @param kerberosDescriptorName kerebros descriptor name
+   * @return information regarding the deleted kerberos descriptor
+   */
+  @DELETE
+  @Path("{kerberosDescriptorName}")
+  @Produces("text/plain")
+  public Response deleteKerberosDescriptor(@Context HttpHeaders headers, @Context UriInfo ui,
+                                           @PathParam("kerberosDescriptorName") String kerberosDescriptorName) {
+    return handleRequest(headers, null, ui, Request.Type.DELETE, createKerberosDescriptorResource(kerberosDescriptorName));
+  }
+
+  private ResourceInstance createKerberosDescriptorResource(String kerberosDescriptorName) {
+    return createResource(Resource.Type.KerberosDescriptor,
+        Collections.singletonMap(Resource.Type.KerberosDescriptor, kerberosDescriptorName));
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/controller/ControllerModule.java

@@ -6,9 +6,9 @@
  * 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
- *
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
  * 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.
@@ -18,33 +18,17 @@
 
 package org.apache.ambari.server.controller;
 
-import static org.eclipse.persistence.config.PersistenceUnitProperties.CREATE_JDBC_DDL_FILE;
-import static org.eclipse.persistence.config.PersistenceUnitProperties.CREATE_ONLY;
-import static org.eclipse.persistence.config.PersistenceUnitProperties.CREATE_OR_EXTEND;
-import static org.eclipse.persistence.config.PersistenceUnitProperties.DDL_BOTH_GENERATION;
-import static org.eclipse.persistence.config.PersistenceUnitProperties.DDL_GENERATION;
-import static org.eclipse.persistence.config.PersistenceUnitProperties.DDL_GENERATION_MODE;
-import static org.eclipse.persistence.config.PersistenceUnitProperties.DROP_AND_CREATE;
-import static org.eclipse.persistence.config.PersistenceUnitProperties.DROP_JDBC_DDL_FILE;
-import static org.eclipse.persistence.config.PersistenceUnitProperties.JDBC_DRIVER;
-import static org.eclipse.persistence.config.PersistenceUnitProperties.JDBC_PASSWORD;
-import static org.eclipse.persistence.config.PersistenceUnitProperties.JDBC_URL;
-import static org.eclipse.persistence.config.PersistenceUnitProperties.JDBC_USER;
-import static org.eclipse.persistence.config.PersistenceUnitProperties.NON_JTA_DATASOURCE;
-import static org.eclipse.persistence.config.PersistenceUnitProperties.THROW_EXCEPTIONS;
-
-import java.beans.PropertyVetoException;
-import java.lang.annotation.Annotation;
-import java.security.SecureRandom;
-import java.text.MessageFormat;
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Properties;
-import java.util.Set;
-
+import com.google.common.util.concurrent.AbstractScheduledService;
+import com.google.common.util.concurrent.ServiceManager;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.inject.AbstractModule;
+import com.google.inject.Scopes;
+import com.google.inject.assistedinject.FactoryModuleBuilder;
+import com.google.inject.name.Names;
+import com.google.inject.persist.PersistModule;
+import com.google.inject.persist.jpa.AmbariJpaPersistModule;
+import com.mchange.v2.c3p0.ComboPooledDataSource;
 import org.apache.ambari.server.AmbariService;
 import org.apache.ambari.server.EagerSingleton;
 import org.apache.ambari.server.StaticallyInject;
@@ -66,6 +50,7 @@ import org.apache.ambari.server.controller.internal.CredentialResourceProvider;
 import org.apache.ambari.server.controller.internal.HostComponentResourceProvider;
 import org.apache.ambari.server.controller.internal.HostKerberosIdentityResourceProvider;
 import org.apache.ambari.server.controller.internal.HostResourceProvider;
+import org.apache.ambari.server.controller.internal.KerberosDescriptorResourceProvider;
 import org.apache.ambari.server.controller.internal.MemberResourceProvider;
 import org.apache.ambari.server.controller.internal.RepositoryVersionResourceProvider;
 import org.apache.ambari.server.controller.internal.ServiceResourceProvider;
@@ -134,17 +119,32 @@ import org.springframework.security.crypto.password.StandardPasswordEncoder;
 import org.springframework.util.ClassUtils;
 import org.springframework.web.filter.DelegatingFilterProxy;
 
-import com.google.common.util.concurrent.AbstractScheduledService;
-import com.google.common.util.concurrent.ServiceManager;
-import com.google.gson.Gson;
-import com.google.gson.GsonBuilder;
-import com.google.inject.AbstractModule;
-import com.google.inject.Scopes;
-import com.google.inject.assistedinject.FactoryModuleBuilder;
-import com.google.inject.name.Names;
-import com.google.inject.persist.PersistModule;
-import com.google.inject.persist.jpa.AmbariJpaPersistModule;
-import com.mchange.v2.c3p0.ComboPooledDataSource;
+import java.beans.PropertyVetoException;
+import java.lang.annotation.Annotation;
+import java.security.SecureRandom;
+import java.text.MessageFormat;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.Set;
+
+import static org.eclipse.persistence.config.PersistenceUnitProperties.CREATE_JDBC_DDL_FILE;
+import static org.eclipse.persistence.config.PersistenceUnitProperties.CREATE_ONLY;
+import static org.eclipse.persistence.config.PersistenceUnitProperties.CREATE_OR_EXTEND;
+import static org.eclipse.persistence.config.PersistenceUnitProperties.DDL_BOTH_GENERATION;
+import static org.eclipse.persistence.config.PersistenceUnitProperties.DDL_GENERATION;
+import static org.eclipse.persistence.config.PersistenceUnitProperties.DDL_GENERATION_MODE;
+import static org.eclipse.persistence.config.PersistenceUnitProperties.DROP_AND_CREATE;
+import static org.eclipse.persistence.config.PersistenceUnitProperties.DROP_JDBC_DDL_FILE;
+import static org.eclipse.persistence.config.PersistenceUnitProperties.JDBC_DRIVER;
+import static org.eclipse.persistence.config.PersistenceUnitProperties.JDBC_PASSWORD;
+import static org.eclipse.persistence.config.PersistenceUnitProperties.JDBC_URL;
+import static org.eclipse.persistence.config.PersistenceUnitProperties.JDBC_USER;
+import static org.eclipse.persistence.config.PersistenceUnitProperties.NON_JTA_DATASOURCE;
+import static org.eclipse.persistence.config.PersistenceUnitProperties.THROW_EXCEPTIONS;
 
 /**
  * Used for injection purposes.
@@ -406,6 +406,7 @@ public class ControllerModule extends AbstractModule {
         .implement(ResourceProvider.class, Names.named("repositoryVersion"), RepositoryVersionResourceProvider.class)
         .implement(ResourceProvider.class, Names.named("hostKerberosIdentity"), HostKerberosIdentityResourceProvider.class)
         .implement(ResourceProvider.class, Names.named("credential"), CredentialResourceProvider.class)
+        .implement(ResourceProvider.class, Names.named("kerberosDescriptor"), KerberosDescriptorResourceProvider.class)
         .build(ResourceProviderFactory.class));
 
     install(new FactoryModuleBuilder().implement(

ambari-server/src/main/java/org/apache/ambari/server/controller/ResourceProviderFactory.java

@@ -22,10 +22,12 @@ package org.apache.ambari.server.controller;
 import java.util.Map;
 import java.util.Set;
 
+import org.apache.ambari.server.controller.spi.Resource;
 import org.apache.ambari.server.controller.spi.Resource.Type;
 import org.apache.ambari.server.controller.spi.ResourceProvider;
 
 import com.google.inject.name.Named;
+import java.util.Set;
 
 public interface ResourceProviderFactory {
   @Named("host")
@@ -61,4 +63,10 @@ public interface ResourceProviderFactory {
 
   @Named("repositoryVersion")
   ResourceProvider getRepositoryVersionResourceProvider();
+
+  @Named("kerberosDescriptor")
+  ResourceProvider getKerberosDescriptorResourceProvider(AmbariManagementController managementController,
+                                                         Set<String> propertyIds,
+                                                         Map<Resource.Type, String> keyPropertyIds);
+
 }

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/AbstractControllerResourceProvider.java

@@ -147,6 +147,8 @@ public abstract class AbstractControllerResourceProvider extends AbstractResourc
         return new HostComponentProcessResourceProvider(propertyIds, keyPropertyIds, managementController);
       case Blueprint:
         return new BlueprintResourceProvider(propertyIds, keyPropertyIds, managementController);
+      case KerberosDescriptor:
+        return resourceProviderFactory.getKerberosDescriptorResourceProvider(managementController, propertyIds, keyPropertyIds);
       case Recommendation:
         return new RecommendationResourceProvider(propertyIds, keyPropertyIds, managementController);
       case Validation:

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/KerberosDescriptorResourceProvider.java

@@ -0,0 +1,182 @@
+package org.apache.ambari.server.controller.internal;
+
+import com.google.inject.assistedinject.Assisted;
+import org.apache.ambari.server.controller.AmbariManagementController;
+import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
+import org.apache.ambari.server.controller.spi.NoSuchResourceException;
+import org.apache.ambari.server.controller.spi.Predicate;
+import org.apache.ambari.server.controller.spi.Request;
+import org.apache.ambari.server.controller.spi.RequestStatus;
+import org.apache.ambari.server.controller.spi.Resource;
+import org.apache.ambari.server.controller.spi.ResourceAlreadyExistsException;
+import org.apache.ambari.server.controller.spi.SystemException;
+import org.apache.ambari.server.controller.spi.UnsupportedPropertyException;
+import org.apache.ambari.server.controller.utilities.PropertyHelper;
+import org.apache.ambari.server.orm.dao.KerberosDescriptorDAO;
+import org.apache.ambari.server.orm.entities.KerberosDescriptorEntity;
+import org.apache.ambari.server.topology.KerberosDescriptor;
+import org.apache.ambari.server.topology.KerberosDescriptorFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Inject;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.
+ */
+public class KerberosDescriptorResourceProvider extends AbstractControllerResourceProvider {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(KerberosDescriptorResourceProvider.class);
+
+  private static final String KERBEROS_DESCRIPTOR_NAME_PROPERTY_ID =
+      PropertyHelper.getPropertyId("KerberosDescriptors", "kerberos_descriptor_name");
+
+  private static final String KERBEROS_DESCRIPTOR_TEXT_PROPERTY_ID =
+      PropertyHelper.getPropertyId("KerberosDescriptors", "kerberos_descriptor_text");
+
+  private KerberosDescriptorDAO kerberosDescriptorDAO;
+
+  private KerberosDescriptorFactory kerberosDescriptorFactory;
+
+  // keep constructors hidden
+  @Inject
+  KerberosDescriptorResourceProvider(KerberosDescriptorDAO kerberosDescriptorDAO,
+                                     KerberosDescriptorFactory kerberosDescriptorFactory,
+                                     @Assisted Set<String> propertyIds,
+                                     @Assisted Map<Resource.Type, String> keyPropertyIds,
+                                     @Assisted AmbariManagementController managementController) {
+    super(propertyIds, keyPropertyIds, managementController);
+    this.kerberosDescriptorDAO = kerberosDescriptorDAO;
+    this.kerberosDescriptorFactory = kerberosDescriptorFactory;
+  }
+
+  @Override
+  public RequestStatus createResources(Request request) throws SystemException, UnsupportedPropertyException,
+      ResourceAlreadyExistsException, NoSuchParentResourceException {
+
+    String name = getNameFromRequest(request);
+    String descriptor = getRawKerberosDescriptorFromRequest(request);
+
+    KerberosDescriptor kerberosDescriptor = kerberosDescriptorFactory.createKerberosDescriptor(name, descriptor);
+    kerberosDescriptorDAO.create(kerberosDescriptor.toEntity());
+
+    return getRequestStatus(null);
+  }
+
+
+  @Override
+  public Set<Resource> getResources(Request request, Predicate predicate) throws SystemException,
+      UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
+
+    List<KerberosDescriptorEntity> results = null;
+    boolean applyPredicate = false;
+
+    if (predicate != null) {
+      Set<Map<String, Object>> requestProps = getPropertyMaps(predicate);
+      if (requestProps.size() == 1) {
+        String name = (String) requestProps.iterator().next().get(
+            KERBEROS_DESCRIPTOR_NAME_PROPERTY_ID);
+
+        if (name != null) {
+          KerberosDescriptorEntity entity = kerberosDescriptorDAO.findByName(name);
+          results = entity == null ? Collections.<KerberosDescriptorEntity>emptyList() :
+              Collections.singletonList(entity);
+        }
+      }
+    }
+
+    if (results == null) {
+      applyPredicate = true;
+      results = kerberosDescriptorDAO.findAll();
+    }
+
+    Set<Resource> resources = new HashSet<Resource>();
+    Set<String> requestPropertyIds = getRequestPropertyIds(request, predicate);
+    for (KerberosDescriptorEntity entity : results) {
+      Resource resource = new ResourceImpl(Resource.Type.KerberosDescriptor);
+      toResource(resource, entity, requestPropertyIds);
+
+      if (predicate == null || !applyPredicate || predicate.evaluate(resource)) {
+        resources.add(resource);
+      }
+    }
+
+    if (predicate != null && resources.isEmpty()) {
+      throw new NoSuchResourceException(
+          "The requested resource doesn't exist: Kerberos Descriptor not found, " + predicate);
+    }
+
+    return resources;
+  }
+
+  private void toResource(Resource resource, KerberosDescriptorEntity entity, Set<String> requestPropertyIds) {
+    setResourceProperty(resource, KERBEROS_DESCRIPTOR_NAME_PROPERTY_ID, entity.getName(), requestPropertyIds);
+    setResourceProperty(resource, KERBEROS_DESCRIPTOR_TEXT_PROPERTY_ID, entity.getKerberosDescriptorText(), requestPropertyIds);
+  }
+
+  @Override
+  public RequestStatus updateResources(Request request, Predicate predicate) throws SystemException,
+      UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
+    throw new UnsupportedOperationException("Not yet implemented!");
+  }
+
+  @Override
+  public RequestStatus deleteResources(Predicate predicate) throws SystemException,
+      UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
+
+    Set<Resource> setResources = getResources(new RequestImpl(null, null, null, null), predicate);
+
+    for (Resource resource : setResources) {
+      final String kerberosDescriptorName =
+          (String) resource.getPropertyValue(KERBEROS_DESCRIPTOR_NAME_PROPERTY_ID);
+      LOGGER.debug("Deleting resource with name: {}", kerberosDescriptorName);
+      kerberosDescriptorDAO.removeByName(kerberosDescriptorName);
+    }
+
+    return getRequestStatus(null);
+  }
+
+  @Override
+  protected Set<String> getPKPropertyIds() {
+    return Collections.emptySet();
+  }
+
+  private String getRawKerberosDescriptorFromRequest(Request request) throws UnsupportedPropertyException {
+    if (request.getRequestInfoProperties() == null ||
+        !request.getRequestInfoProperties().containsKey(Request.REQUEST_INFO_BODY_PROPERTY)) {
+      LOGGER.error("Could not find the raw request body in the request: {}", request);
+      throw new UnsupportedPropertyException(Resource.Type.KerberosDescriptor,
+          Collections.singleton(Request.REQUEST_INFO_BODY_PROPERTY));
+    }
+    return request.getRequestInfoProperties().get(Request.REQUEST_INFO_BODY_PROPERTY);
+  }
+
+  private String getNameFromRequest(Request request) throws UnsupportedPropertyException {
+    if (request.getProperties() == null || !request.getProperties().iterator().hasNext()) {
+      LOGGER.error("There is no {} property id in the request {}", KERBEROS_DESCRIPTOR_NAME_PROPERTY_ID, request);
+      throw new UnsupportedPropertyException(Resource.Type.KerberosDescriptor,
+          Collections.singleton(KERBEROS_DESCRIPTOR_NAME_PROPERTY_ID));
+    }
+    return (String) request.getProperties().iterator().next().get(KERBEROS_DESCRIPTOR_NAME_PROPERTY_ID);
+  }
+
+}

ambari-server/src/main/java/org/apache/ambari/server/controller/spi/Resource.java

@@ -6,9 +6,9 @@
  * 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
- *
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
  * 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.
@@ -145,7 +145,8 @@ public interface Resource {
     ActiveWidgetLayout,
     Theme,
     HostKerberosIdentity,
-    Credential;
+    Credential,
+    KerberosDescriptor;
 
     /**
      * Get the {@link Type} that corresponds to this InternalType.
@@ -252,6 +253,7 @@ public interface Resource {
     public static final Type ActiveWidgetLayout = InternalType.ActiveWidgetLayout.getType();
     public static final Type HostKerberosIdentity = InternalType.HostKerberosIdentity.getType();
     public static final Type Credential = InternalType.Credential.getType();
+    public static final Type KerberosDescriptor = InternalType.KerberosDescriptor.getType();
 
     /**
      * The type name.

ambari-server/src/main/java/org/apache/ambari/server/orm/dao/KerberosDescriptorDAO.java

@@ -0,0 +1,107 @@
+package org.apache.ambari.server.orm.dao;
+
+import com.google.inject.persist.Transactional;
+import org.apache.ambari.server.orm.RequiresSession;
+import org.apache.ambari.server.orm.entities.KerberosDescriptorEntity;
+
+import javax.inject.Inject;
+import javax.inject.Provider;
+import javax.inject.Singleton;
+import javax.persistence.EntityManager;
+import javax.persistence.TypedQuery;
+import java.util.List;
+
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.
+ */
+
+/**
+ *
+ */
+@Singleton
+public class KerberosDescriptorDAO {
+
+  @Inject
+  private Provider<EntityManager> entityManagerProvider;
+
+  @RequiresSession
+  public KerberosDescriptorEntity findByName(String kerberosDescriptorName) {
+    return entityManagerProvider.get().find(KerberosDescriptorEntity.class, kerberosDescriptorName);
+  }
+
+  /**
+   * Find all kerberos descriptors.
+   *
+   * @return all kerberos descriptors or an empty List
+   */
+  @RequiresSession
+  public List<KerberosDescriptorEntity> findAll() {
+    TypedQuery<KerberosDescriptorEntity> query = entityManagerProvider.get().
+            createNamedQuery("allKerberosDescriptors", KerberosDescriptorEntity.class);
+    return query.getResultList();
+  }
+
+  @Transactional
+  public void create(KerberosDescriptorEntity kerberosDescriptorEntity) {
+    entityManagerProvider.get().persist(kerberosDescriptorEntity);
+  }
+
+
+  /**
+   * Refresh the state of the instance from the database,
+   * overwriting changes made to the entity, if any.
+   *
+   * @param kerberosDescriptorEntity entity to refresh
+   */
+  @Transactional
+  public void refresh(KerberosDescriptorEntity kerberosDescriptorEntity) {
+    entityManagerProvider.get().refresh(kerberosDescriptorEntity);
+  }
+
+  /**
+   * Merge the state of the given entity into the current persistence context.
+   *
+   * @param kerberosDescriptorEntity entity to merge
+   * @return the merged entity
+   */
+  @Transactional
+  public KerberosDescriptorEntity merge(KerberosDescriptorEntity kerberosDescriptorEntity) {
+    return entityManagerProvider.get().merge(kerberosDescriptorEntity);
+  }
+
+  /**
+   * Remove the entity instance.
+   *
+   * @param kerberosDescriptorEntity entity to remove
+   */
+  @Transactional
+  public void remove(KerberosDescriptorEntity kerberosDescriptorEntity) {
+    entityManagerProvider.get().remove(merge(kerberosDescriptorEntity));
+  }
+
+  /**
+   * Remove entity instance by primary key
+   *
+   * @param name Primary key: kerberos descriptor name
+   */
+  @Transactional
+  public void removeByName(String name) {
+    entityManagerProvider.get().remove(findByName(name));
+  }
+
+
+}

ambari-server/src/main/java/org/apache/ambari/server/orm/entities/KerberosDescriptorEntity.java

@@ -0,0 +1,56 @@
+package org.apache.ambari.server.orm.entities;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.NamedQuery;
+import javax.persistence.Table;
+
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.
+ */
+@Table(name = "kerberos_descriptor")
+@Entity
+@NamedQuery(name = "allKerberosDescriptors",
+        query = "SELECT kerberosDescriptor  FROM KerberosDescriptorEntity kerberosDescriptor")
+public class KerberosDescriptorEntity {
+
+  @Id
+  @Column(name = "kerberos_descriptor_name", nullable = false, insertable = true, updatable = false,
+          unique = true, length = 100)
+  private String name;
+
+  @Column(name = "kerberos_descriptor", nullable = false, insertable = true, updatable = false)
+  private String kerberosDescriptorText;
+
+
+  public String getName() {
+    return name;
+  }
+
+  public void setName(String name) {
+    this.name = name;
+  }
+
+  public String getKerberosDescriptorText() {
+    return kerberosDescriptorText;
+  }
+
+  public void setKerberosDescriptorText(String kerberosDescriptorText) {
+    this.kerberosDescriptorText = kerberosDescriptorText;
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/topology/KerberosDescriptor.java

@@ -0,0 +1,26 @@
+package org.apache.ambari.server.topology;
+
+import org.apache.ambari.server.orm.entities.KerberosDescriptorEntity;
+
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.
+ */
+public interface KerberosDescriptor {
+    String getName();
+
+    KerberosDescriptorEntity toEntity();
+}

ambari-server/src/main/java/org/apache/ambari/server/topology/KerberosDescriptorFactory.java

@@ -0,0 +1,30 @@
+package org.apache.ambari.server.topology;
+
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.
+ */
+
+/**
+ * Factory for kerberos descriptor instances
+ */
+public class KerberosDescriptorFactory {
+
+    public KerberosDescriptor createKerberosDescriptor(String name, String descriptor) {
+        KerberosDescriptor kerberosDescriptor = new KerberosDescriptorImpl(name, descriptor);
+        return kerberosDescriptor;
+    }
+}

ambari-server/src/main/java/org/apache/ambari/server/topology/KerberosDescriptorImpl.java

@@ -0,0 +1,50 @@
+package org.apache.ambari.server.topology;
+
+import org.apache.ambari.server.orm.entities.KerberosDescriptorEntity;
+
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.
+ */
+public class KerberosDescriptorImpl implements KerberosDescriptor {
+
+    private final String name;
+
+    private final String descriptor;
+
+
+    public KerberosDescriptorImpl(String name, String descriptor) {
+        this.name = name;
+        this.descriptor = descriptor;
+    }
+
+    @Override
+    public String getName() {
+        return name;
+    }
+
+    public String getDescriptor() {
+        return descriptor;
+    }
+
+    @Override
+    public KerberosDescriptorEntity toEntity() {
+        KerberosDescriptorEntity entity = new KerberosDescriptorEntity();
+        entity.setName(getName());
+        entity.setKerberosDescriptorText(getDescriptor());
+        return entity;
+    }
+}

ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog213.java

@@ -24,6 +24,7 @@ import com.google.inject.Inject;
 import com.google.inject.Injector;
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.controller.AmbariManagementController;
+import org.apache.ambari.server.orm.DBAccessor;
 import org.apache.ambari.server.orm.dao.AlertDefinitionDAO;
 import org.apache.ambari.server.orm.dao.DaoUtils;
 import org.apache.ambari.server.orm.entities.AlertDefinitionEntity;
@@ -37,9 +38,11 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.sql.SQLException;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
@@ -57,12 +60,16 @@ public class UpgradeCatalog213 extends AbstractUpgradeCatalog {
   private static final String HBASE_ENV_CONFIG = "hbase-env";
   private static final String CONTENT_PROPERTY = "content";
 
+  private static final String KERBEROS_DESCRIPTOR_TABLE = "kerberos_descriptor";
+  private static final String KERBEROS_DESCRIPTOR_NAME_COLUMN = "kerberos_descriptor_name";
+  private static final String KERBEROS_DESCRIPTOR_COLUMN = "kerberos_descriptor";
 
   /**
    * Logger.
    */
   private static final Logger LOG = LoggerFactory.getLogger(UpgradeCatalog213.class);
 
+
   @Inject
   DaoUtils daoUtils;
 
@@ -106,6 +113,17 @@ public class UpgradeCatalog213 extends AbstractUpgradeCatalog {
    */
   @Override
   protected void executeDDLUpdates() throws AmbariException, SQLException {
+    addKerberosDescriptorTable();
+  }
+
+  private void addKerberosDescriptorTable() throws SQLException {
+
+    List<DBAccessor.DBColumnInfo> columns = new ArrayList<DBAccessor.DBColumnInfo>();
+    columns.add(new DBAccessor.DBColumnInfo(KERBEROS_DESCRIPTOR_NAME_COLUMN, String.class, 255, null, false));
+    columns.add(new DBAccessor.DBColumnInfo(KERBEROS_DESCRIPTOR_COLUMN, char[].class, null, null, false));
+
+    LOG.debug("Creating table [ {} ] with columns [ {} ] and primary key: [ {} ]", KERBEROS_DESCRIPTOR_TABLE, columns, KERBEROS_DESCRIPTOR_NAME_COLUMN);
+    dbAccessor.createTable(KERBEROS_DESCRIPTOR_TABLE, columns, KERBEROS_DESCRIPTOR_NAME_COLUMN);
   }
 
   /**

ambari-server/src/main/resources/Ambari-DDL-MySQL-CREATE.sql

@@ -758,6 +758,13 @@ CREATE TABLE kerberos_principal_host (
   PRIMARY KEY(principal_name, host_id)
 );
 
+CREATE TABLE kerberos_descriptor
+(
+   kerberos_descriptor_name   VARCHAR(255) NOT NULL,
+   kerberos_descriptor        TEXT NOT NULL,
+   PRIMARY KEY (kerberos_descriptor_name)
+);
+
 ALTER TABLE kerberos_principal_host ADD CONSTRAINT FK_krb_pr_host_id FOREIGN KEY (host_id) REFERENCES hosts (host_id);
 ALTER TABLE kerberos_principal_host ADD CONSTRAINT FK_krb_pr_host_principalname FOREIGN KEY (principal_name) REFERENCES kerberos_principal (principal_name);
 -- Kerberos (end)

ambari-server/src/main/resources/Ambari-DDL-Oracle-CREATE.sql

@@ -747,6 +747,13 @@ CREATE TABLE kerberos_principal_host (
   PRIMARY KEY(principal_name, host_id)
 );
 
+CREATE TABLE kerberos_descriptor
+(
+   kerberos_descriptor_name   VARCHAR2(255) NOT NULL,
+   kerberos_descriptor        CLOB NOT NULL,
+   PRIMARY KEY (kerberos_descriptor_name)
+);
+
 ALTER TABLE kerberos_principal_host ADD CONSTRAINT FK_krb_pr_host_id FOREIGN KEY (host_id) REFERENCES hosts (host_id);
 ALTER TABLE kerberos_principal_host ADD CONSTRAINT FK_krb_pr_host_principalname FOREIGN KEY (principal_name) REFERENCES kerberos_principal (principal_name);
 -- Kerberos (end)

ambari-server/src/main/resources/Ambari-DDL-Postgres-CREATE.sql

@@ -749,6 +749,13 @@ CREATE TABLE kerberos_principal_host (
   PRIMARY KEY(principal_name, host_id)
 );
 
+CREATE TABLE kerberos_descriptor
+(
+   kerberos_descriptor_name   VARCHAR(255) NOT NULL,
+   kerberos_descriptor        TEXT NOT NULL,
+   PRIMARY KEY (kerberos_descriptor_name)
+);
+
 ALTER TABLE kerberos_principal_host ADD CONSTRAINT FK_krb_pr_host_id FOREIGN KEY (host_id) REFERENCES hosts (host_id);
 ALTER TABLE kerberos_principal_host ADD CONSTRAINT FK_krb_pr_host_principalname FOREIGN KEY (principal_name) REFERENCES kerberos_principal (principal_name);
 -- Kerberos (end)

ambari-server/src/main/resources/Ambari-DDL-Postgres-EMBEDDED-CREATE.sql

@@ -830,6 +830,14 @@ CREATE TABLE ambari.kerberos_principal_host (
 );
 GRANT ALL PRIVILEGES ON TABLE ambari.kerberos_principal_host TO :username;
 
+CREATE TABLE ambari.kerberos_descriptor
+(
+   kerberos_descriptor_name   VARCHAR(255) NOT NULL,
+   kerberos_descriptor        TEXT NOT NULL,
+   PRIMARY KEY (kerberos_descriptor_name)
+);
+GRANT ALL PRIVILEGES ON TABLE ambari.kerberos_descriptor TO :username;
+
 ALTER TABLE ambari.kerberos_principal_host ADD CONSTRAINT FK_krb_pr_host_id FOREIGN KEY (host_id) REFERENCES ambari.hosts (host_id);
 ALTER TABLE ambari.kerberos_principal_host ADD CONSTRAINT FK_krb_pr_host_principalname FOREIGN KEY (principal_name) REFERENCES ambari.kerberos_principal (principal_name);
 -- Kerberos (end)

ambari-server/src/main/resources/Ambari-DDL-SQLAnywhere-CREATE.sql

@@ -748,6 +748,13 @@ CREATE TABLE kerberos_principal_host (
   PRIMARY KEY(principal_name, host_id)
 );
 
+CREATE TABLE kerberos_descriptor
+(
+   kerberos_descriptor_name   VARCHAR(255) NOT NULL,
+   kerberos_descriptor        TEXT NOT NULL,
+   PRIMARY KEY (kerberos_descriptor_name)
+);
+
 ALTER TABLE kerberos_principal_host ADD CONSTRAINT FK_krb_pr_host_id FOREIGN KEY (host_id) REFERENCES hosts (host_id);
 ALTER TABLE kerberos_principal_host ADD CONSTRAINT FK_krb_pr_host_principalname FOREIGN KEY (principal_name) REFERENCES kerberos_principal (principal_name);
 -- Kerberos (end)

ambari-server/src/main/resources/Ambari-DDL-SQLServer-CREATE.sql

@@ -864,6 +864,13 @@ CREATE TABLE kerberos_principal_host (
   PRIMARY KEY CLUSTERED (principal_name, host_id)
 );
 
+CREATE TABLE kerberos_descriptor
+(
+   kerberos_descriptor_name   VARCHAR(255) NOT NULL,
+   kerberos_descriptor        VARCHAR(MAX) NOT NULL,
+   PRIMARY KEY (kerberos_descriptor_name)
+);
+
 ALTER TABLE kerberos_principal_host ADD CONSTRAINT FK_krb_pr_host_id FOREIGN KEY (host_id) REFERENCES hosts (host_id);
 ALTER TABLE kerberos_principal_host ADD CONSTRAINT FK_krb_pr_host_principalname FOREIGN KEY (principal_name) REFERENCES kerberos_principal (principal_name);
 -- Kerberos (end)

ambari-server/src/main/resources/META-INF/persistence.xml

@@ -87,6 +87,7 @@
     <class>org.apache.ambari.server.orm.entities.TopologyHostInfoEntity</class>
     <class>org.apache.ambari.server.orm.entities.TopologyHostTaskEntity</class>
     <class>org.apache.ambari.server.orm.entities.TopologyLogicalTaskEntity</class>
+    <class>org.apache.ambari.server.orm.entities.KerberosDescriptorEntity</class>
 
     <properties>      
       <property name="eclipselink.cache.size.default" value="10000" />

ambari-server/src/main/resources/key_properties.json

@@ -147,5 +147,8 @@
     "Stack": "StackLevelConfigurations/stack_name",
     "StackVersion": "StackLevelConfigurations/stack_version",
     "StackLevelConfiguration": "StackLevelConfigurations/property_name"  
+  },
+  "KerberosDescriptor": {
+    "KerberosDescriptor": "KerberosDescriptors/kerberos_descriptor_name"
   }
 }

ambari-server/src/main/resources/properties.json

@@ -455,5 +455,11 @@
         "StackConfigurationDependency/dependency_type",
         "StackConfigurationDependency/dependency_name",
         "_"
-    ]
+    ],
+  "KerberosDescriptor":[
+    "KerberosDescriptors/kerberos_descriptor_name",
+    "KerberosDescriptors/kerberos_descriptor_text",
+    "_"
+  ]
+
 }

ambari-server/src/test/java/org/apache/ambari/server/controller/internal/KerberosDescriptorResourceProviderTest.java

@@ -0,0 +1,160 @@
+package org.apache.ambari.server.controller.internal;
+
+import org.apache.ambari.server.controller.spi.Request;
+import org.apache.ambari.server.controller.spi.UnsupportedPropertyException;
+import org.apache.ambari.server.orm.dao.KerberosDescriptorDAO;
+import org.apache.ambari.server.orm.entities.KerberosDescriptorEntity;
+import org.apache.ambari.server.topology.KerberosDescriptorFactory;
+import org.apache.ambari.server.topology.KerberosDescriptorImpl;
+import org.easymock.Capture;
+import org.easymock.EasyMock;
+import org.easymock.EasyMockRule;
+import org.easymock.Mock;
+import org.easymock.MockType;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import static org.easymock.EasyMock.anyString;
+import static org.easymock.EasyMock.capture;
+import static org.easymock.EasyMock.reset;
+
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.
+ */
+
+public class KerberosDescriptorResourceProviderTest {
+
+  private static final String TEST_KERBEROS_DESCRIPTOR_NAME = "descriptor-name-0";
+  private static final String TEST_KERBEROS_DESCRIPTOR = "descriptor";
+  public static final String KERBEROS_DESCRIPTORS_KERBEROS_DESCRIPTOR_NAME = "KerberosDescriptors/kerberos_descriptor_name";
+
+  @Rule
+  public EasyMockRule mocks = new EasyMockRule(this);
+
+  @Mock(type = MockType.STRICT)
+  private KerberosDescriptorDAO kerberosDescriptorDAO;
+
+  @Mock(type = MockType.STRICT)
+  private KerberosDescriptorFactory kerberosDescriptorFactory;
+
+  @Mock(type = MockType.STRICT)
+  private Request request;
+
+  private KerberosDescriptorResourceProvider kerberosDescriptorResourceProvider;
+
+  @Before
+  public void before() {
+    reset(request);
+
+  }
+
+  @Test(expected = UnsupportedPropertyException.class)
+  public void testCreateShouldThrowExceptionWhenNoDescriptorProvided() throws Exception {
+
+    // GIVEN
+    EasyMock.expect(request.getProperties()).andReturn(requestPropertySet(KERBEROS_DESCRIPTORS_KERBEROS_DESCRIPTOR_NAME,
+        TEST_KERBEROS_DESCRIPTOR_NAME)).times(3);
+    EasyMock.expect(request.getRequestInfoProperties()).andReturn(requestInfoPropertyMap("", "")).times(2);
+    EasyMock.replay(request);
+
+    kerberosDescriptorResourceProvider = new KerberosDescriptorResourceProvider(kerberosDescriptorDAO,
+        kerberosDescriptorFactory, Collections.EMPTY_SET, Collections.EMPTY_MAP, null);
+
+    // WHEN
+    kerberosDescriptorResourceProvider.createResources(request);
+
+    // THEN
+    // exception is thrown
+  }
+
+  @Test(expected = UnsupportedPropertyException.class)
+  public void testCreateShouldThrowExceptionWhenNoNameProvided() throws Exception {
+
+    // GIVEN
+    EasyMock.expect(request.getProperties()).andReturn(emptyRequestPropertySet()).times(2);
+    EasyMock.replay(request);
+
+    kerberosDescriptorResourceProvider = new KerberosDescriptorResourceProvider(kerberosDescriptorDAO,
+        kerberosDescriptorFactory, Collections.EMPTY_SET, Collections.EMPTY_MAP, null);
+
+    // WHEN
+    kerberosDescriptorResourceProvider.createResources(request);
+
+    // THEN
+    // exception is thrown
+  }
+
+
+  @Test
+  public void testShoudCreateResourceWhenNameAndDescriptorProvided() throws Exception {
+
+    // GIVEN
+    kerberosDescriptorResourceProvider = new KerberosDescriptorResourceProvider(kerberosDescriptorDAO,
+        kerberosDescriptorFactory, Collections.EMPTY_SET, Collections.EMPTY_MAP, null);
+
+    EasyMock.expect(request.getProperties())
+        .andReturn(requestPropertySet(KERBEROS_DESCRIPTORS_KERBEROS_DESCRIPTOR_NAME, TEST_KERBEROS_DESCRIPTOR_NAME))
+        .times(3);
+    EasyMock.expect(request.getRequestInfoProperties())
+        .andReturn(requestInfoPropertyMap(Request.REQUEST_INFO_BODY_PROPERTY, TEST_KERBEROS_DESCRIPTOR))
+        .times(3);
+    EasyMock.expect(kerberosDescriptorFactory.createKerberosDescriptor(anyString(), anyString()))
+        .andReturn(new KerberosDescriptorImpl(TEST_KERBEROS_DESCRIPTOR_NAME, TEST_KERBEROS_DESCRIPTOR));
+
+    Capture<KerberosDescriptorEntity> entityCapture = EasyMock.newCapture();
+    kerberosDescriptorDAO.create(capture(entityCapture));
+
+    EasyMock.replay(request, kerberosDescriptorFactory, kerberosDescriptorDAO);
+
+    // WHEN
+    kerberosDescriptorResourceProvider.createResources(request);
+
+    // THEN
+    Assert.assertNotNull(entityCapture.getValue());
+    Assert.assertEquals("The resource name is invalid!", TEST_KERBEROS_DESCRIPTOR_NAME, entityCapture.getValue()
+        .getName());
+
+  }
+
+  private Set<Map<String, Object>> emptyRequestPropertySet() {
+    return Collections.emptySet();
+  }
+
+
+  private Map<String, String> requestInfoPropertyMap(String propertyKey, String propertyValue) {
+    Map<String, String> propsMap = new HashMap<>();
+    propsMap.put(propertyKey, propertyValue);
+    return propsMap;
+  }
+
+  private Set<Map<String, Object>> requestPropertySet(String testPropertyKey, String testPropertyValue) {
+    Set<Map<String, Object>> invalidProps = new HashSet<>();
+    Map<String, Object> invalidMap = new HashMap<>();
+    invalidMap.put(testPropertyKey, testPropertyValue);
+    invalidProps.add(invalidMap);
+    return invalidProps;
+  }
+
+}

ambari-server/src/test/java/org/apache/ambari/server/orm/dao/KerberosDescriptorDAOTest.java

@@ -0,0 +1,82 @@
+package org.apache.ambari.server.orm.dao;
+
+import com.google.inject.Provider;
+import org.apache.ambari.server.orm.entities.KerberosDescriptorEntity;
+import org.easymock.Capture;
+import org.easymock.EasyMock;
+import org.easymock.EasyMockRule;
+import org.easymock.Mock;
+import org.easymock.MockType;
+import org.easymock.TestSubject;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+import javax.persistence.EntityManager;
+
+import static org.easymock.EasyMock.capture;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.reset;
+
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.
+ */
+public class KerberosDescriptorDAOTest {
+
+  public static final String TEST_KERBEROS_DESCRIPTOR_ENTITY_NAME = "test-kerberos-descriptor-entity";
+
+  @Rule
+  public EasyMockRule mocks = new EasyMockRule(this);
+
+  @Mock(type = MockType.STRICT)
+  private Provider<EntityManager> entityManagerProvider;
+
+  @Mock(type = MockType.STRICT)
+  private EntityManager entityManager;
+
+  @TestSubject
+  private KerberosDescriptorDAO kerberosDescriptorDAO = new KerberosDescriptorDAO();
+
+  @Before
+  public void before() {
+    reset(entityManagerProvider);
+    expect(entityManagerProvider.get()).andReturn(entityManager).atLeastOnce();
+    replay(entityManagerProvider);
+  }
+
+
+  @Test
+  public void testPersistNewKerberosDescriptorEntity() {
+    //GIVEN
+    KerberosDescriptorEntity kerberosDescriptorEntity = new KerberosDescriptorEntity();
+    kerberosDescriptorEntity.setName(TEST_KERBEROS_DESCRIPTOR_ENTITY_NAME);
+
+    Capture<KerberosDescriptorEntity> capturedArgument = EasyMock.newCapture();
+    entityManager.persist(capture(capturedArgument));
+    replay(entityManager);
+
+    //WHEN
+    kerberosDescriptorDAO.create(kerberosDescriptorEntity);
+
+    //THEN
+    Assert.assertNotNull(capturedArgument);
+    Assert.assertEquals("The persisted object is not the expected one", kerberosDescriptorEntity, capturedArgument.getValue());
+  }
+
+}

ambari-server/src/test/java/org/apache/ambari/server/scheduler/ExecutionScheduleManagerTest.java

@@ -6,9 +6,9 @@
  * 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
- *
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
  * 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.
@@ -17,29 +17,16 @@
  */
 package org.apache.ambari.server.scheduler;
 
-import static org.easymock.EasyMock.anyObject;
-import static org.easymock.EasyMock.capture;
-import static org.easymock.EasyMock.createMock;
-import static org.easymock.EasyMock.createMockBuilder;
-import static org.easymock.EasyMock.createNiceMock;
-import static org.easymock.EasyMock.eq;
-import static org.easymock.EasyMock.expect;
-import static org.easymock.EasyMock.expectLastCall;
-import static org.easymock.EasyMock.replay;
-import static org.easymock.EasyMock.verify;
-import static org.hamcrest.CoreMatchers.instanceOf;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-
+import com.google.gson.Gson;
+import com.google.inject.Binder;
+import com.google.inject.Guice;
+import com.google.inject.Inject;
+import com.google.inject.Injector;
+import com.google.inject.Module;
+import com.google.inject.persist.PersistService;
+import com.google.inject.persist.Transactional;
+import com.google.inject.util.Modules;
 import junit.framework.Assert;
-
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.actionmanager.ActionDBAccessor;
 import org.apache.ambari.server.actionmanager.HostRoleStatus;
@@ -60,6 +47,7 @@ import org.apache.ambari.server.state.scheduler.RequestExecution;
 import org.apache.ambari.server.state.scheduler.RequestExecutionFactory;
 import org.apache.ambari.server.state.scheduler.Schedule;
 import org.easymock.Capture;
+import org.easymock.EasyMock;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -78,15 +66,26 @@ import org.quartz.impl.matchers.GroupMatcher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.google.gson.Gson;
-import com.google.inject.Binder;
-import com.google.inject.Guice;
-import com.google.inject.Inject;
-import com.google.inject.Injector;
-import com.google.inject.Module;
-import com.google.inject.persist.PersistService;
-import com.google.inject.persist.Transactional;
-import com.google.inject.util.Modules;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.easymock.EasyMock.anyObject;
+import static org.easymock.EasyMock.capture;
+import static org.easymock.EasyMock.createMock;
+import static org.easymock.EasyMock.createMockBuilder;
+import static org.easymock.EasyMock.createNiceMock;
+import static org.easymock.EasyMock.eq;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.expectLastCall;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
 
 public class ExecutionScheduleManagerTest {
   private Clusters clusters;
@@ -365,13 +364,13 @@ public class ExecutionScheduleManagerTest {
     batchRequestResponse.setRequestId(requestId);
     batchRequestResponse.setReturnCode(202);
 
-    ExecutionScheduleManager scheduleManager = createMockBuilder(ExecutionScheduleManager.class).
-      withConstructor(configurationMock, executionSchedulerMock, tokenStorageMock, clustersMock,
-        actionDBAccessorMock, gson).
-      addMockedMethods("performApiRequest", "updateBatchRequest").createNiceMock();
+    EasyMock.expect(configurationMock.getApiSSLAuthentication()).andReturn(Boolean.FALSE);
+    EasyMock.replay(configurationMock);
 
-    //interesting easymock behavior, workaround to not to expect method called in constructor
-    expectLastCall().anyTimes();
+    ExecutionScheduleManager scheduleManager = createMockBuilder(ExecutionScheduleManager.class).
+        withConstructor(configurationMock, executionSchedulerMock, tokenStorageMock, clustersMock,
+            actionDBAccessorMock, gson).
+        addMockedMethods("performApiRequest", "updateBatchRequest").createNiceMock();
 
     expect(clustersMock.getCluster(clusterName)).andReturn(clusterMock).anyTimes();
     expect(clusterMock.getAllRequestExecutions()).andReturn(executionMap).anyTimes();
@@ -391,13 +390,13 @@ public class ExecutionScheduleManagerTest {
     actionDBAccessorMock.setSourceScheduleForRequest(eq(requestId), eq(executionId));
     expectLastCall().once();
 
-    replay(clusterMock, clustersMock, configurationMock, requestExecutionMock, executionSchedulerMock,
-      tokenStorageMock, batchRequestMock, scheduleManager, actionDBAccessorMock);
+    replay(clusterMock, clustersMock, requestExecutionMock, executionSchedulerMock,
+        tokenStorageMock, batchRequestMock, scheduleManager, actionDBAccessorMock);
 
     scheduleManager.executeBatchRequest(executionId, batchId, clusterName);
 
     verify(clusterMock, clustersMock, configurationMock, requestExecutionMock, executionSchedulerMock,
-      tokenStorageMock, batchRequestMock, scheduleManager, actionDBAccessorMock);
+        tokenStorageMock, batchRequestMock, scheduleManager, actionDBAccessorMock);
 
   }
 
@@ -426,13 +425,13 @@ public class ExecutionScheduleManagerTest {
     batchRequestResponse.setRequestId(requestId);
     batchRequestResponse.setReturnCode(202);
 
-    ExecutionScheduleManager scheduleManager = createMockBuilder(ExecutionScheduleManager.class).
-      withConstructor(configurationMock, executionSchedulerMock, tokenStorageMock, clustersMock,
-        actionDBAccessorMock, gson).
-      addMockedMethods("performApiRequest").createNiceMock();
+    EasyMock.expect(configurationMock.getApiSSLAuthentication()).andReturn(Boolean.FALSE);
+    EasyMock.replay(configurationMock);
 
-    //interesting easymock behavior, workaround to not to expect method called in constructor
-    expectLastCall().anyTimes();
+    ExecutionScheduleManager scheduleManager = createMockBuilder(ExecutionScheduleManager.class).
+        withConstructor(configurationMock, executionSchedulerMock, tokenStorageMock, clustersMock,
+            actionDBAccessorMock, gson).
+        addMockedMethods("performApiRequest").createNiceMock();
 
     expect(clustersMock.getCluster(clusterName)).andReturn(clusterMock).anyTimes();
     expect(clusterMock.getAllRequestExecutions()).andReturn(executionMap).anyTimes();
@@ -441,13 +440,13 @@ public class ExecutionScheduleManagerTest {
     expectLastCall().once();
 
 
-    replay(clusterMock, clustersMock, configurationMock, requestExecutionMock, executionSchedulerMock,
-      tokenStorageMock, batchRequestMock, scheduleManager);
+    replay(clusterMock, clustersMock, requestExecutionMock, executionSchedulerMock,
+        tokenStorageMock, batchRequestMock, scheduleManager);
 
     scheduleManager.updateBatchRequest(executionId, batchId, clusterName, batchRequestResponse, true);
 
     verify(clusterMock, clustersMock, configurationMock, requestExecutionMock, executionSchedulerMock,
-      tokenStorageMock, batchRequestMock, scheduleManager);
+        tokenStorageMock, batchRequestMock, scheduleManager);
 
   }
 
@@ -464,31 +463,30 @@ public class ExecutionScheduleManagerTest {
     long requestId = 5L;
     String clusterName = "mycluster";
     String apiUri = "api/v1/clusters/mycluster/requests/5";
-    Capture<String> uriCapture= new Capture<String>();
+    Capture<String> uriCapture = new Capture<String>();
 
     BatchRequestResponse batchRequestResponse = new BatchRequestResponse();
     batchRequestResponse.setStatus(HostRoleStatus.IN_PROGRESS.toString());
     batchRequestResponse.setRequestId(requestId);
     batchRequestResponse.setReturnCode(202);
 
-    ExecutionScheduleManager scheduleManager = createMockBuilder(ExecutionScheduleManager.class).
-      withConstructor(configurationMock, executionSchedulerMock, tokenStorageMock, clustersMock,
-        actionDBAccessorMock, gson).
-      addMockedMethods("performApiGetRequest").createNiceMock();
-
-    //interesting easymock behavior, workaround to not to expect method called in constructor
-    expectLastCall().anyTimes();
+    EasyMock.expect(configurationMock.getApiSSLAuthentication()).andReturn(Boolean.FALSE);
+    EasyMock.replay(configurationMock);
 
+    ExecutionScheduleManager scheduleManager = createMockBuilder(ExecutionScheduleManager.class).
+        withConstructor(configurationMock, executionSchedulerMock, tokenStorageMock, clustersMock,
+            actionDBAccessorMock, gson).
+        addMockedMethods("performApiGetRequest").createNiceMock();
 
     expect(scheduleManager.performApiGetRequest(capture(uriCapture), eq(true))).andReturn(batchRequestResponse).once();
 
-    replay(clusterMock, clustersMock, configurationMock, executionSchedulerMock,
-      tokenStorageMock, scheduleManager);
+    replay(clusterMock, clustersMock, executionSchedulerMock,
+        tokenStorageMock, scheduleManager);
 
     scheduleManager.getBatchRequestResponse(requestId, clusterName);
 
     verify(clusterMock, clustersMock, configurationMock, executionSchedulerMock,
-      tokenStorageMock, scheduleManager);
+        tokenStorageMock, scheduleManager);
 
     assertEquals(apiUri, uriCapture.getValue());
   }
@@ -520,11 +518,11 @@ public class ExecutionScheduleManagerTest {
     expect(batchMock.getBatchSettings()).andReturn(batchSettings).anyTimes();
 
     replay(clustersMock, clusterMock, configurationMock, requestExecutionMock,
-      executionSchedulerMock, batchMock);
+        executionSchedulerMock, batchMock);
 
     ExecutionScheduleManager scheduleManager =
-      new ExecutionScheduleManager(configurationMock, executionSchedulerMock,
-        tokenStorageMock, clustersMock, actionDBAccessorMock, gson);
+        new ExecutionScheduleManager(configurationMock, executionSchedulerMock,
+            tokenStorageMock, clustersMock, actionDBAccessorMock, gson);
 
     HashMap<String, Integer> taskCounts = new HashMap<String, Integer>() {{
       put(BatchRequestJob.BATCH_REQUEST_FAILED_TASKS_KEY, 2);
@@ -564,50 +562,51 @@ public class ExecutionScheduleManagerTest {
     Map<Long, RequestExecution> executionMap = new HashMap<Long, RequestExecution>();
     executionMap.put(executionId, requestExecutionMock);
 
-    ExecutionScheduleManager scheduleManager =
-      createMockBuilder(ExecutionScheduleManager.class).withConstructor
-        (configurationMock, executionSchedulerMock, tokenStorageMock,
-          clustersMock, actionDBAccessorMock, gson).createMock();
+    EasyMock.expect(configurationMock.getApiSSLAuthentication()).andReturn(Boolean.FALSE);
+    EasyMock.replay(configurationMock);
 
-    expectLastCall().anyTimes();
+    ExecutionScheduleManager scheduleManager =
+        createMockBuilder(ExecutionScheduleManager.class)
+            .withConstructor(configurationMock, executionSchedulerMock, tokenStorageMock,
+                clustersMock, actionDBAccessorMock, gson).createMock();
 
     expect(clustersMock.getCluster(clusterName)).andReturn(clusterMock).anyTimes();
     expect(clusterMock.getAllRequestExecutions()).andReturn(executionMap).anyTimes();
     expect(requestExecutionMock.getBatch()).andReturn(batchMock).anyTimes();
     expect(batchMock.getBatchRequests()).andReturn
-      (new ArrayList<BatchRequest>() {{
-        add(batchRequestMock);
-      }});
+        (new ArrayList<BatchRequest>() {{
+          add(batchRequestMock);
+        }});
     expect(batchRequestMock.getOrderId()).andReturn(1L).anyTimes();
     expect(executionSchedulerMock.getJobDetail((JobKey) anyObject()))
-      .andReturn(jobDetailMock).anyTimes();
+        .andReturn(jobDetailMock).anyTimes();
     expect((List<Trigger>) executionSchedulerMock
-      .getTriggersForJob((JobKey) anyObject())).andReturn(triggers).anyTimes();
+        .getTriggersForJob((JobKey) anyObject())).andReturn(triggers).anyTimes();
     expect(triggerMock.mayFireAgain()).andReturn(true).anyTimes();
     expect(triggerMock.getFinalFireTime()).andReturn(pastDate).anyTimes();
 
     requestExecutionMock.updateStatus(RequestExecution.Status.COMPLETED);
     expectLastCall();
 
-    replay(clustersMock, clusterMock, configurationMock, requestExecutionMock,
-      executionSchedulerMock, scheduleManager, batchMock, batchRequestMock,
-      triggerMock, jobDetailMock, actionDBAccessorMock);
+    replay(clustersMock, clusterMock, requestExecutionMock,
+        executionSchedulerMock, scheduleManager, batchMock, batchRequestMock,
+        triggerMock, jobDetailMock, actionDBAccessorMock);
 
     scheduleManager.finalizeBatch(executionId, clusterName);
 
     verify(clustersMock, clusterMock, configurationMock, requestExecutionMock,
-      executionSchedulerMock, scheduleManager, batchMock, batchRequestMock,
-      triggerMock, jobDetailMock, actionDBAccessorMock);
+        executionSchedulerMock, scheduleManager, batchMock, batchRequestMock,
+        triggerMock, jobDetailMock, actionDBAccessorMock);
   }
 
   @Test
   public void testFinalizeBeforeExit() throws Exception {
     ExecutionScheduleManager scheduleManagerMock = createMock(ExecutionScheduleManager.class);
     AbstractLinearExecutionJob executionJob =
-      createMockBuilder(AbstractLinearExecutionJob.class)
-      .addMockedMethods("finalizeExecution", "doWork")
-      .withConstructor(scheduleManagerMock)
-      .createMock();
+        createMockBuilder(AbstractLinearExecutionJob.class)
+            .addMockedMethods("finalizeExecution", "doWork")
+            .withConstructor(scheduleManagerMock)
+            .createMock();
     JobExecutionContext context = createMock(JobExecutionContext.class);
     JobDetail jobDetail = createMock(JobDetail.class);
     JobDataMap jobDataMap = createMock(JobDataMap.class);

ambari-server/src/test/java/org/apache/ambari/server/upgrade/UpgradeCatalog213Test.java

@@ -27,10 +27,12 @@ import com.google.inject.Provider;
 import com.google.inject.persist.PersistService;
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.api.services.AmbariMetaInfo;
+import org.apache.ambari.server.configuration.Configuration;
 import org.apache.ambari.server.controller.AmbariManagementController;
 import org.apache.ambari.server.orm.DBAccessor;
 import org.apache.ambari.server.orm.GuiceJpaInitializer;
 import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
+import org.apache.ambari.server.orm.dao.DaoUtils;
 import org.apache.ambari.server.orm.dao.StackDAO;
 import org.apache.ambari.server.orm.entities.StackEntity;
 import org.apache.ambari.server.state.Cluster;
@@ -40,7 +42,10 @@ import org.apache.ambari.server.state.ConfigHelper;
 import org.apache.ambari.server.state.Service;
 import org.apache.ambari.server.state.StackId;
 import org.apache.ambari.server.state.stack.OsFamily;
+import org.easymock.Capture;
+import org.easymock.EasyMock;
 import org.easymock.EasyMockSupport;
+import org.easymock.IMocksControl;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -50,9 +55,11 @@ import javax.persistence.EntityManager;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.HashMap;
-import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 
+import static org.easymock.EasyMock.anyObject;
+import static org.easymock.EasyMock.capture;
 import static org.easymock.EasyMock.createMockBuilder;
 import static org.easymock.EasyMock.createNiceMock;
 import static org.easymock.EasyMock.createStrictMock;
@@ -72,6 +79,8 @@ public class UpgradeCatalog213Test {
   private UpgradeCatalogHelper upgradeCatalogHelper;
   private StackEntity desiredStackEntity;
 
+  private IMocksControl mocksControl = EasyMock.createControl();
+
   @Before
   public void init() {
     reset(entityManagerProvider);
@@ -211,7 +220,7 @@ public class UpgradeCatalog213Test {
     expect(mockClusters.getClusters()).andReturn(new HashMap<String, Cluster>() {{
       put("normal", mockClusterExpected);
     }}).atLeastOnce();
-    expect(mockClusterExpected.getCurrentStackVersion()).andReturn(new StackId("HDP","2.2"));
+    expect(mockClusterExpected.getCurrentStackVersion()).andReturn(new StackId("HDP", "2.2"));
 
     expect(mockClusterExpected.getDesiredConfigByType("hbase-env")).andReturn(mockHbaseEnv).atLeastOnce();
     expect(mockHbaseEnv.getProperties()).andReturn(propertiesHbaseEnv).atLeastOnce();
@@ -225,7 +234,7 @@ public class UpgradeCatalog213Test {
   @Test
   public void testUpdateHDFSConfiguration() throws Exception {
     EasyMockSupport easyMockSupport = new EasyMockSupport();
-    final AmbariManagementController  mockAmbariManagementController = easyMockSupport.createNiceMock(AmbariManagementController.class);
+    final AmbariManagementController mockAmbariManagementController = easyMockSupport.createNiceMock(AmbariManagementController.class);
     final ConfigHelper mockConfigHelper = easyMockSupport.createMock(ConfigHelper.class);
 
     final Clusters mockClusters = easyMockSupport.createStrictMock(Clusters.class);
@@ -269,29 +278,29 @@ public class UpgradeCatalog213Test {
     Method updateAmsHbaseEnvContent = UpgradeCatalog213.class.getDeclaredMethod("updateAmsHbaseEnvContent", String.class);
     UpgradeCatalog213 upgradeCatalog213 = new UpgradeCatalog213(injector);
     String oldContent = "export HBASE_CLASSPATH=${HBASE_CLASSPATH}\n" +
-            "\n" +
-            "# The maximum amount of heap to use, in MB. Default is 1000.\n" +
-            "export HBASE_HEAPSIZE={{hbase_heapsize}}\n" +
-            "\n" +
-            "{% if java_version &lt; 8 %}\n" +
-            "export HBASE_MASTER_OPTS=\" -XX:PermSize=64m -XX:MaxPermSize={{hbase_master_maxperm_size}} -Xms{{hbase_heapsize}} -Xmx{{hbase_heapsize}} -Xmn{{hbase_master_xmn_size}} -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly\"\n" +
-            "export HBASE_REGIONSERVER_OPTS=\"-XX:MaxPermSize=128m -Xmn{{regionserver_xmn_size}} -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly -Xms{{regionserver_heapsize}} -Xmx{{regionserver_heapsize}}\"\n" +
-            "{% else %}\n" +
-            "export HBASE_MASTER_OPTS=\" -Xms{{hbase_heapsize}} -Xmx{{hbase_heapsize}} -Xmn{{hbase_master_xmn_size}} -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly\"\n" +
-            "export HBASE_REGIONSERVER_OPTS=\" -Xmn{{regionserver_xmn_size}} -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly -Xms{{regionserver_heapsize}} -Xmx{{regionserver_heapsize}}\"\n" +
-            "{% endif %}\n";
+        "\n" +
+        "# The maximum amount of heap to use, in MB. Default is 1000.\n" +
+        "export HBASE_HEAPSIZE={{hbase_heapsize}}\n" +
+        "\n" +
+        "{% if java_version &lt; 8 %}\n" +
+        "export HBASE_MASTER_OPTS=\" -XX:PermSize=64m -XX:MaxPermSize={{hbase_master_maxperm_size}} -Xms{{hbase_heapsize}} -Xmx{{hbase_heapsize}} -Xmn{{hbase_master_xmn_size}} -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly\"\n" +
+        "export HBASE_REGIONSERVER_OPTS=\"-XX:MaxPermSize=128m -Xmn{{regionserver_xmn_size}} -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly -Xms{{regionserver_heapsize}} -Xmx{{regionserver_heapsize}}\"\n" +
+        "{% else %}\n" +
+        "export HBASE_MASTER_OPTS=\" -Xms{{hbase_heapsize}} -Xmx{{hbase_heapsize}} -Xmn{{hbase_master_xmn_size}} -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly\"\n" +
+        "export HBASE_REGIONSERVER_OPTS=\" -Xmn{{regionserver_xmn_size}} -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly -Xms{{regionserver_heapsize}} -Xmx{{regionserver_heapsize}}\"\n" +
+        "{% endif %}\n";
     String expectedContent = "export HBASE_CLASSPATH=${HBASE_CLASSPATH}\n" +
-            "\n" +
-            "# The maximum amount of heap to use, in MB. Default is 1000.\n" +
-            "export HBASE_HEAPSIZE={{hbase_heapsize}}m\n" +
-            "\n" +
-            "{% if java_version &lt; 8 %}\n" +
-            "export HBASE_MASTER_OPTS=\" -XX:PermSize=64m -XX:MaxPermSize={{hbase_master_maxperm_size}}m -Xms{{hbase_heapsize}}m -Xmx{{hbase_heapsize}}m -Xmn{{hbase_master_xmn_size}}m -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly\"\n" +
-            "export HBASE_REGIONSERVER_OPTS=\"-XX:MaxPermSize=128m -Xmn{{regionserver_xmn_size}}m -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly -Xms{{regionserver_heapsize}}m -Xmx{{regionserver_heapsize}}m\"\n" +
-            "{% else %}\n" +
-            "export HBASE_MASTER_OPTS=\" -Xms{{hbase_heapsize}}m -Xmx{{hbase_heapsize}}m -Xmn{{hbase_master_xmn_size}}m -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly\"\n" +
-            "export HBASE_REGIONSERVER_OPTS=\" -Xmn{{regionserver_xmn_size}}m -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly -Xms{{regionserver_heapsize}}m -Xmx{{regionserver_heapsize}}m\"\n" +
-            "{% endif %}\n";
+        "\n" +
+        "# The maximum amount of heap to use, in MB. Default is 1000.\n" +
+        "export HBASE_HEAPSIZE={{hbase_heapsize}}m\n" +
+        "\n" +
+        "{% if java_version &lt; 8 %}\n" +
+        "export HBASE_MASTER_OPTS=\" -XX:PermSize=64m -XX:MaxPermSize={{hbase_master_maxperm_size}}m -Xms{{hbase_heapsize}}m -Xmx{{hbase_heapsize}}m -Xmn{{hbase_master_xmn_size}}m -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly\"\n" +
+        "export HBASE_REGIONSERVER_OPTS=\"-XX:MaxPermSize=128m -Xmn{{regionserver_xmn_size}}m -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly -Xms{{regionserver_heapsize}}m -Xmx{{regionserver_heapsize}}m\"\n" +
+        "{% else %}\n" +
+        "export HBASE_MASTER_OPTS=\" -Xms{{hbase_heapsize}}m -Xmx{{hbase_heapsize}}m -Xmn{{hbase_master_xmn_size}}m -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly\"\n" +
+        "export HBASE_REGIONSERVER_OPTS=\" -Xmn{{regionserver_xmn_size}}m -XX:CMSInitiatingOccupancyFraction=70 -XX:+UseCMSInitiatingOccupancyOnly -Xms{{regionserver_heapsize}}m -Xmx{{regionserver_heapsize}}m\"\n" +
+        "{% endif %}\n";
     String result = (String) updateAmsHbaseEnvContent.invoke(upgradeCatalog213, oldContent);
     Assert.assertEquals(expectedContent, result);
   }
@@ -301,9 +310,9 @@ public class UpgradeCatalog213Test {
     Method updateAmsEnvContent = UpgradeCatalog213.class.getDeclaredMethod("updateAmsEnvContent", String.class);
     UpgradeCatalog213 upgradeCatalog213 = new UpgradeCatalog213(injector);
     String oldContent = "# AMS Collector heapsize\n" +
-            "export AMS_COLLECTOR_HEAPSIZE={{metrics_collector_heapsize}}\n";
+        "export AMS_COLLECTOR_HEAPSIZE={{metrics_collector_heapsize}}\n";
     String expectedContent = "# AMS Collector heapsize\n" +
-            "export AMS_COLLECTOR_HEAPSIZE={{metrics_collector_heapsize}}m\n";
+        "export AMS_COLLECTOR_HEAPSIZE={{metrics_collector_heapsize}}m\n";
     String result = (String) updateAmsEnvContent.invoke(upgradeCatalog213, oldContent);
     Assert.assertEquals(expectedContent, result);
   }
@@ -365,18 +374,18 @@ public class UpgradeCatalog213Test {
   public void testModifyJournalnodeProcessAlertSource() throws Exception {
     UpgradeCatalog213 upgradeCatalog213 = new UpgradeCatalog213(injector);
     String alertSource = "{\"uri\":\"{{hdfs-site/dfs.journalnode.http-address}}\",\"default_port\":8480," +
-            "\"type\":\"PORT\",\"reporting\":{\"ok\":{\"text\":\"TCP OK - {0:.3f}s response on port {1}\"}," +
-            "\"warning\":{\"text\":\"TCP OK - {0:.3f}s response on port {1}\",\"value\":1.5}," +
-            "\"critical\":{\"text\":\"Connection failed: {0} to {1}:{2}\",\"value\":5.0}}}";
+        "\"type\":\"PORT\",\"reporting\":{\"ok\":{\"text\":\"TCP OK - {0:.3f}s response on port {1}\"}," +
+        "\"warning\":{\"text\":\"TCP OK - {0:.3f}s response on port {1}\",\"value\":1.5}," +
+        "\"critical\":{\"text\":\"Connection failed: {0} to {1}:{2}\",\"value\":5.0}}}";
     String expected = "{\"reporting\":{\"ok\":{\"text\":\"HTTP {0} response in {2:.3f}s\"}," +
-            "\"warning\":{\"text\":\"HTTP {0} response from {1} in {2:.3f}s ({3})\"}," +
-            "\"critical\":{\"text\":\"Connection failed to {1} ({3})\"}},\"type\":\"WEB\"," +
-            "\"uri\":{\"http\":\"{{hdfs-site/dfs.journalnode.http-address}}\"," +
-            "\"https\":\"{{hdfs-site/dfs.journalnode.https-address}}\"," +
-            "\"kerberos_keytab\":\"{{hdfs-site/dfs.web.authentication.kerberos.keytab}}\"," +
-            "\"kerberos_principal\":\"{{hdfs-site/dfs.web.authentication.kerberos.principal}}\"," +
-            "\"https_property\":\"{{hdfs-site/dfs.http.policy}}\"," +
-            "\"https_property_value\":\"HTTPS_ONLY\",\"connection_timeout\":5.0}}";
+        "\"warning\":{\"text\":\"HTTP {0} response from {1} in {2:.3f}s ({3})\"}," +
+        "\"critical\":{\"text\":\"Connection failed to {1} ({3})\"}},\"type\":\"WEB\"," +
+        "\"uri\":{\"http\":\"{{hdfs-site/dfs.journalnode.http-address}}\"," +
+        "\"https\":\"{{hdfs-site/dfs.journalnode.https-address}}\"," +
+        "\"kerberos_keytab\":\"{{hdfs-site/dfs.web.authentication.kerberos.keytab}}\"," +
+        "\"kerberos_principal\":\"{{hdfs-site/dfs.web.authentication.kerberos.principal}}\"," +
+        "\"https_property\":\"{{hdfs-site/dfs.http.policy}}\"," +
+        "\"https_property_value\":\"HTTPS_ONLY\",\"connection_timeout\":5.0}}";
     Assert.assertEquals(expected, upgradeCatalog213.modifyJournalnodeProcessAlertSource(alertSource));
   }
 
@@ -460,4 +469,48 @@ public class UpgradeCatalog213Test {
 
     Assert.assertEquals("2.1.3", upgradeCatalog.getTargetVersion());
   }
+
+  @Test
+  public void testShouldDDLsBeExecutedOnUpgrade() throws Exception {
+    // GIVEN
+    Injector mockedInjector = mocksControl.createMock(Injector.class);
+    DBAccessor mockedDbAccessor = mocksControl.createMock(DBAccessor.class);
+    DaoUtils mockedDaoUtils = mocksControl.createMock(DaoUtils.class);
+    Configuration mockedConfiguration = mocksControl.createMock(Configuration.class);
+    StackUpgradeUtil mockedStackUpgradeUtil = mocksControl.createMock(StackUpgradeUtil.class);
+
+    Capture<String> capturedTableName = EasyMock.newCapture();
+    Capture<String> capturedPKColumn = EasyMock.newCapture();
+    Capture<List<DBAccessor.DBColumnInfo>> capturedColumns = EasyMock.newCapture();
+
+    EasyMock.expect(mockedInjector.getInstance(DaoUtils.class)).andReturn(mockedDaoUtils);
+    mockedInjector.injectMembers(anyObject(UpgradeCatalog.class));
+    EasyMock.expect(mockedConfiguration.getDatabaseType()).andReturn(Configuration.DatabaseType.POSTGRES).times(2);
+    EasyMock.expect(mockedConfiguration.getDatabaseUser()).andReturn("ambari");
+    EasyMock.expect(mockedConfiguration.getServerJDBCPostgresSchemaName()).andReturn("fo");
+
+    mockedDbAccessor.executeQuery("ALTER SCHEMA fo OWNER TO \"ambari\";");
+    mockedDbAccessor.executeQuery("ALTER ROLE \"ambari\" SET search_path to 'fo';");
+    mockedDbAccessor.createTable(capture(capturedTableName), capture(capturedColumns), capture(capturedPKColumn));
+
+    mocksControl.replay();
+
+    UpgradeCatalog213 testSubject = new UpgradeCatalog213(mockedInjector);
+    EasyMockSupport.injectMocks(testSubject);
+
+    //todo refactor the DI approach, don't directly access these members!!!
+    testSubject.stackUpgradeUtil = mockedStackUpgradeUtil;
+    testSubject.dbAccessor = mockedDbAccessor;
+    testSubject.configuration = mockedConfiguration;
+
+    // WHEN
+    testSubject.upgradeSchema();
+
+    // THEN
+    Assert.assertEquals("The table name is wrong!", "kerberos_descriptor", capturedTableName.getValue());
+    Assert.assertEquals("The primary key is wrong!", "kerberos_descriptor_name", capturedPKColumn.getValue());
+    Assert.assertTrue("Ther number of columns is wrong!", capturedColumns.getValue().size() == 2);
+
+  }
+
 }