content: /**
   * 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.
   */
  package org.apache.atlas;
  
  import com.fasterxml.jackson.databind.JsonNode;
  import com.fasterxml.jackson.databind.node.ObjectNode;
  import com.fasterxml.jackson.jaxrs.json.JacksonJaxbJsonProvider;
  import com.google.common.annotations.VisibleForTesting;
  import com.sun.jersey.api.client.Client;
  import com.sun.jersey.api.client.ClientHandlerException;
  import com.sun.jersey.api.client.ClientResponse;
  import com.sun.jersey.api.client.GenericType;
  import com.sun.jersey.api.client.WebResource;
  import com.sun.jersey.api.client.config.DefaultClientConfig;
  import com.sun.jersey.api.client.filter.HTTPBasicAuthFilter;
  import com.sun.jersey.api.json.JSONConfiguration;
  import com.sun.jersey.client.urlconnection.URLConnectionClientHandler;
  import com.sun.jersey.multipart.BodyPart;
  import com.sun.jersey.multipart.FormDataBodyPart;
  import com.sun.jersey.multipart.FormDataMultiPart;
  import com.sun.jersey.multipart.MultiPart;
  import com.sun.jersey.multipart.file.FileDataBodyPart;
  import com.sun.jersey.multipart.file.StreamDataBodyPart;
  import com.sun.jersey.multipart.impl.MultiPartWriter;
  import org.apache.atlas.model.impexp.AtlasExportRequest;
  import org.apache.atlas.model.impexp.AtlasImportRequest;
  import org.apache.atlas.model.impexp.AtlasImportResult;
  import org.apache.atlas.model.impexp.AtlasServer;
  import org.apache.atlas.model.metrics.AtlasMetrics;
  import org.apache.atlas.security.SecureClientUtils;
  import org.apache.atlas.type.AtlasType;
  import org.apache.atlas.utils.AtlasJson;
  import org.apache.atlas.utils.AuthenticationUtil;
  import org.apache.commons.configuration.Configuration;
  import org.apache.commons.io.IOUtils;
  import org.apache.commons.lang.StringUtils;
  import org.apache.hadoop.security.UserGroupInformation;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import javax.ws.rs.HttpMethod;
  import javax.ws.rs.core.Cookie;
  import javax.ws.rs.core.MediaType;
  import javax.ws.rs.core.MultivaluedMap;
  import javax.ws.rs.core.Response;
  import javax.ws.rs.core.UriBuilder;
  
  import java.io.File;
  import java.io.FileOutputStream;
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.OutputStream;
  import java.net.ConnectException;
  import java.net.URI;
  import java.util.List;
  import java.util.Map;
  
  import static org.apache.atlas.security.SecurityProperties.TLS_ENABLED;
  
  public abstract class AtlasBaseClient {
      private static final Logger LOG = LoggerFactory.getLogger(AtlasBaseClient.class);
  
      public static final String BASE_URI              = "api/atlas/";
      public static final String TYPES                 = "types";
      public static final String ADMIN_VERSION         = "admin/version";
      public static final String ADMIN_STATUS          = "admin/status";
      public static final String ADMIN_METRICS         = "admin/metrics";
      public static final String ADMIN_IMPORT          = "admin/import";
      public static final String ADMIN_EXPORT          = "admin/export";
      public static final String ADMIN_SERVER_TEMPLATE = "%sadmin/server/%s";
  
      public static final String QUERY  = "query";
      public static final String LIMIT  = "limit";
      public static final String OFFSET = "offset";
      public static final String STATUS = "Status";
  
      public static final API API_STATUS  = new API(BASE_URI + ADMIN_STATUS, HttpMethod.GET, Response.Status.OK);
      public static final API API_VERSION = new API(BASE_URI + ADMIN_VERSION, HttpMethod.GET, Response.Status.OK);
      public static final API API_METRICS = new API(BASE_URI + ADMIN_METRICS, HttpMethod.GET, Response.Status.OK);
  
      static final String JSON_MEDIA_TYPE             = MediaType.APPLICATION_JSON + "; charset=UTF-8";
      static final String UNKNOWN_STATUS              = "Unknown status";
      static final String ATLAS_CLIENT_HA_RETRIES_KEY = "atlas.client.ha.retries";
  
      // Setting the default value based on testing failovers while client code like quickstart is running.
      static final int    DEFAULT_NUM_RETRIES                   = 4;
      static final String ATLAS_CLIENT_HA_SLEEP_INTERVAL_MS_KEY = "atlas.client.ha.sleep.interval.ms";
  
      // Setting the default value based on testing failovers while client code like quickstart is running.
      // With number of retries, this gives a total time of about 20s for the server to start.
      static final int    DEFAULT_SLEEP_BETWEEN_RETRIES_MS      = 5000;
  
      private static final API    IMPORT                  = new API(BASE_URI + ADMIN_IMPORT, HttpMethod.POST, Response.Status.OK, MediaType.MULTIPART_FORM_DATA, MediaType.APPLICATION_JSON);
      private static final API    EXPORT                  = new API(BASE_URI + ADMIN_EXPORT, HttpMethod.POST, Response.Status.OK, MediaType.APPLICATION_JSON, MediaType.APPLICATION_OCTET_STREAM);
      private static final String IMPORT_REQUEST_PARAMTER = "request";
      private static final String IMPORT_DATA_PARAMETER   = "data";
  
      protected WebResource        service;
      protected Configuration      configuration;
      private   String             basicAuthUser;
      private   String             basicAuthPassword;
      private   AtlasClientContext atlasClientContext;
      private   boolean            retryEnabled;
      private   Cookie             cookie;
  
      private SecureClientUtils clientUtils;
  
      protected AtlasBaseClient() {
      }
  
      protected AtlasBaseClient(String[] baseUrl, String[] basicAuthUserNamePassword) {
          if (basicAuthUserNamePassword != null) {
              if (basicAuthUserNamePassword.length > 0) {
                  this.basicAuthUser = basicAuthUserNamePassword[0];
              }
  
              if (basicAuthUserNamePassword.length > 1) {
                  this.basicAuthPassword = basicAuthUserNamePassword[1];
              }
          }
  
          initializeState(baseUrl, null, null);
      }
  
      protected AtlasBaseClient(String... baseUrls) throws AtlasException {
          this(getCurrentUGI(), baseUrls);
      }
  
      protected AtlasBaseClient(UserGroupInformation ugi, String[] baseUrls) {
          this(ugi, ugi.getShortUserName(), baseUrls);
      }
  
      protected AtlasBaseClient(UserGroupInformation ugi, String doAsUser, String[] baseUrls) {
          initializeState(baseUrls, ugi, doAsUser);
      }
  
      protected AtlasBaseClient(String[] baseUrls, Cookie cookie) {
          this.cookie = cookie;
  
          initializeState(baseUrls, null, null);
      }
  
      @VisibleForTesting
      protected AtlasBaseClient(WebResource service, Configuration configuration) {
          this.service       = service;
          this.configuration = configuration;
      }
  
      @VisibleForTesting
      protected AtlasBaseClient(Configuration configuration, String[] baseUrl, String[] basicAuthUserNamePassword) {
          if (basicAuthUserNamePassword != null) {
              if (basicAuthUserNamePassword.length > 0) {
                  this.basicAuthUser = basicAuthUserNamePassword[0];
              }
  
              if (basicAuthUserNamePassword.length > 1) {
                  this.basicAuthPassword = basicAuthUserNamePassword[1];
              }
          }
  
          initializeState(configuration, baseUrl, null, null);
      }
  
      protected static UserGroupInformation getCurrentUGI() throws AtlasException {
          try {
              return UserGroupInformation.getCurrentUser();
          } catch (IOException e) {
              throw new AtlasException(e);
          }
      }
  
      public void setCookie(Cookie cookie) {
          this.cookie = cookie;
      }
  
      public boolean isServerReady() throws AtlasServiceException {
          WebResource resource = getResource(API_VERSION.getNormalizedPath());
  
          try {
              callAPIWithResource(API_VERSION, resource, null, ObjectNode.class);
  
              return true;
          } catch (ClientHandlerException che) {
              return false;
          } catch (AtlasServiceException ase) {
              if (ase.getStatus() != null && ase.getStatus().equals(ClientResponse.Status.SERVICE_UNAVAILABLE)) {
                  LOG.warn("Received SERVICE_UNAVAILABLE, server is not yet ready");
  
                  return false;
              }
  
              throw ase;
          }
      }
  
      /**
       * Return status of the service instance the client is pointing to.
       *
       * @return One of the values in ServiceState.ServiceStateValue or {@link #UNKNOWN_STATUS} if
       * there is a JSON parse exception
       * @throws AtlasServiceException if there is a HTTP error.
       */
      public String getAdminStatus() throws AtlasServiceException {
          String      result   = AtlasBaseClient.UNKNOWN_STATUS;
          WebResource resource = getResource(service, API_STATUS.getNormalizedPath());
          ObjectNode  response = callAPIWithResource(API_STATUS, resource, null, ObjectNode.class);
  
          if (response.has(STATUS)) {
              result = response.get(STATUS).asText();
          }
  
          return result;
      }
  
      /**
       * @return Return metrics of the service instance the client is pointing to
       * @throws AtlasServiceException
       */
      public AtlasMetrics getAtlasMetrics() throws AtlasServiceException {
          return callAPI(API_METRICS, AtlasMetrics.class, null);
      }
  
      public <T> T callAPI(API api, Class<T> responseType, Object requestObject, String... params) throws AtlasServiceException {
          return callAPIWithResource(api, getResource(api, params), requestObject, responseType);
      }
  
      public <T> T callAPI(API api, GenericType<T> responseType, Object requestObject, String... params) throws AtlasServiceException {
          return callAPIWithResource(api, getResource(api, params), requestObject, responseType);
      }
  
      public <T> T callAPI(API api, Class<T> responseType, Object requestBody, MultivaluedMap<String, String> queryParams, String... params) throws AtlasServiceException {
          WebResource resource = getResource(api, queryParams, params);
  
          return callAPIWithResource(api, resource, requestBody, responseType);
      }
  
      public <T> T callAPI(API api, Class<T> responseType, MultivaluedMap<String, String> queryParams, String... params) throws AtlasServiceException {
          WebResource resource = getResource(api, queryParams, params);
  
          return callAPIWithResource(api, resource, null, responseType);
      }
  
      public <T> T callAPI(API api, GenericType<T> responseType, MultivaluedMap<String, String> queryParams, String... params) throws AtlasServiceException {
          WebResource resource = getResource(api, queryParams, params);
  
          return callAPIWithResource(api, resource, null, responseType);
      }
  
      public <T> T callAPI(API api, Class<T> responseType, MultivaluedMap<String, String> queryParams) throws AtlasServiceException {
          return callAPIWithResource(api, getResource(api, queryParams), null, responseType);
      }
  
      public <T> T callAPI(API api, Class<T> responseType, String queryParamKey, List<String> queryParamValues) throws AtlasServiceException {
          return callAPIWithResource(api, getResource(api, queryParamKey, queryParamValues), null, responseType);
      }
  
      public void close() {
          if (clientUtils != null) {
              clientUtils.destroyFactory();
          }
      }
  
      public InputStream exportData(AtlasExportRequest request) throws AtlasServiceException {
          try {
              return (InputStream) callAPI(EXPORT, Object.class, request);
          } catch (AtlasServiceException e) {
              LOG.error("error in export API call", e);
  
              throw new AtlasServiceException(e);
          }
      }
  
      public void exportData(AtlasExportRequest request, String absolutePath) throws AtlasServiceException {
          OutputStream fileOutputStream = null;
          InputStream  inputStream      = exportData(request);
  
          try {
              fileOutputStream = new FileOutputStream(new File(absolutePath));
  
              byte[] buffer = new byte[8 * 1024];
              int    bytesRead;
  
              while ((bytesRead = inputStream.read(buffer)) != -1) {
                  fileOutputStream.write(buffer, 0, bytesRead);
              }
  
              IOUtils.closeQuietly(inputStream);
              IOUtils.closeQuietly(fileOutputStream);
          } catch (Exception e) {
              LOG.error("error writing to file", e);
  
              throw new AtlasServiceException(e);
          } finally {
              if (fileOutputStream != null) {
                  try {
                      fileOutputStream.close();
                  } catch (IOException e) {
                      LOG.error("error closing file", e);
  
                      throw new AtlasServiceException(e);
                  }
              }
          }
      }
  
      public AtlasImportResult importData(AtlasImportRequest request, String absoluteFilePath) throws AtlasServiceException {
          return performImportData(getImportRequestBodyPart(request), new FileDataBodyPart(IMPORT_DATA_PARAMETER, new File(absoluteFilePath)));
      }
  
      public AtlasImportResult importData(AtlasImportRequest request, InputStream stream) throws AtlasServiceException {
          return performImportData(getImportRequestBodyPart(request), new StreamDataBodyPart(IMPORT_DATA_PARAMETER, stream));
      }
  
      public AtlasServer getServer(String serverName) throws AtlasServiceException {
          API api = new API(String.format(ADMIN_SERVER_TEMPLATE, BASE_URI, serverName), HttpMethod.GET, Response.Status.OK);
  
          return callAPI(api, AtlasServer.class, null);
      }
  
      @VisibleForTesting
      protected Client getClient(Configuration configuration, UserGroupInformation ugi, String doAsUser) {
          DefaultClientConfig config = new DefaultClientConfig();
  
          // Enable POJO mapping feature
          config.getFeatures().put(JSONConfiguration.FEATURE_POJO_MAPPING, Boolean.TRUE);
          config.getClasses().add(JacksonJaxbJsonProvider.class);
          config.getClasses().add(MultiPartWriter.class);
  
          int readTimeout    = configuration.getInt("atlas.client.readTimeoutMSecs", 60000);
          int connectTimeout = configuration.getInt("atlas.client.connectTimeoutMSecs", 60000);
  
          if (configuration.getBoolean(TLS_ENABLED, false)) {
              // create an SSL properties configuration if one doesn't exist.  SSLFactory expects a file, so forced
              // to create a
              // configuration object, persist it, then subsequently pass in an empty configuration to SSLFactory
              try {
                  SecureClientUtils.persistSSLClientConfiguration(configuration, System.getProperty("atlas.conf"));
              } catch (Exception e) {
                  LOG.info("Error processing client configuration.", e);
              }
          }
  
          clientUtils = new SecureClientUtils();
  
          boolean isKerberosEnabled = AuthenticationUtil.isKerberosAuthenticationEnabled(ugi);
  
          final URLConnectionClientHandler handler;
  
          if (isKerberosEnabled) {
              handler = clientUtils.getClientConnectionHandler(config, configuration, doAsUser, ugi);
          } else {
              if (configuration.getBoolean(TLS_ENABLED, false)) {
                  handler = clientUtils.getUrlConnectionClientHandler();
              } else {
                  handler = new URLConnectionClientHandler();
              }
          }
  
          Client client = new Client(handler, config);
  
          client.setReadTimeout(readTimeout);
          client.setConnectTimeout(connectTimeout);
  
          return client;
      }
  
      @VisibleForTesting
      protected String determineActiveServiceURL(String[] baseUrls, Client client) {
          if (baseUrls.length == 0) {
              throw new IllegalArgumentException("Base URLs cannot be null or empty");
          }
  
          final String        baseUrl;
          AtlasServerEnsemble atlasServerEnsemble = new AtlasServerEnsemble(baseUrls);
  
          if (atlasServerEnsemble.hasSingleInstance()) {
              baseUrl = atlasServerEnsemble.firstURL();
  
              LOG.info("Client has only one service URL, will use that for all actions: {}", baseUrl);
          } else {
              try {
                  baseUrl = selectActiveServerAddress(client, atlasServerEnsemble);
              } catch (AtlasServiceException e) {
                  LOG.error("None of the passed URLs are active: {}", atlasServerEnsemble, e);
  
                  throw new IllegalArgumentException("None of the passed URLs are active " + atlasServerEnsemble, e);
              }
          }
  
          return baseUrl;
      }
  
      protected Configuration getClientProperties() {
          try {
              if (configuration == null) {
                  configuration = ApplicationProperties.get();
              }
          } catch (AtlasException e) {
              LOG.error("Exception while loading configuration.", e);
          }
  
          return configuration;
      }
  
      protected WebResource getResource(String path, String... pathParams) {
          return getResource(service, path, pathParams);
      }
  
      protected <T> T callAPIWithResource(API api, WebResource resource, Object requestObject, Class<T> responseType) throws AtlasServiceException {
          GenericType<T> genericType = null;
  
          if (responseType != null) {
              genericType = new GenericType<>(responseType);
          }
  
          return callAPIWithResource(api, resource, requestObject, genericType);
      }
  
      protected <T> T callAPIWithResource(API api, WebResource resource, Object requestObject, GenericType<T> responseType) throws AtlasServiceException {
          ClientResponse clientResponse;
          int            i = 0;
  
          do {
              if (LOG.isDebugEnabled()) {
                  LOG.debug("------------------------------------------------------");
                  LOG.debug("Call         : {} {}", api.getMethod(), api.getNormalizedPath());
                  LOG.debug("Content-type : {} ", api.getConsumes());
                  LOG.debug("Accept       : {} ", api.getProduces());
                  LOG.debug("Request      : {}", requestObject);
              }
  
              WebResource.Builder requestBuilder = resource.getRequestBuilder();
  
              // Set content headers
              requestBuilder
                      .accept(api.getProduces())
                      .type(api.getConsumes())
                      .header("Expect", "100-continue");
  
              // Set cookie if present
              if (cookie != null) {
                  requestBuilder.cookie(cookie);
              }
  
              clientResponse = requestBuilder.method(api.getMethod(), ClientResponse.class, requestObject);
  
              LOG.debug("HTTP Status  : {}", clientResponse.getStatus());
  
              if (LOG.isInfoEnabled()) {
                  LOG.info("method={} path={} contentType={} accept={} status={}", api.getMethod(), api.getNormalizedPath(), api.getConsumes(), api.getProduces(), clientResponse.getStatus());
              }
  
              if (clientResponse.getStatus() == api.getExpectedStatus().getStatusCode()) {
                  if (responseType == null) {
                      return null;
                  }
  
                  try {
                      if (api.getProduces().equals(MediaType.APPLICATION_OCTET_STREAM)) {
                          return (T) clientResponse.getEntityInputStream();
                      } else if (responseType.getRawClass().equals(ObjectNode.class)) {
                          String stringEntity = clientResponse.getEntity(String.class);
  
                          try {
                              JsonNode jsonObject = AtlasJson.parseToV1JsonNode(stringEntity);
  
                              LOG.debug("Response     : {}", jsonObject);
                              LOG.debug("------------------------------------------------------");
  
                              return (T) jsonObject;
                          } catch (IOException e) {
                              throw new AtlasServiceException(api, e);
                          }
                      } else {
                          T entity = clientResponse.getEntity(responseType);
  
                          LOG.debug("Response     : {}", entity);
                          LOG.debug("------------------------------------------------------");
  
                          return entity;
                      }
                  } catch (ClientHandlerException e) {
                      throw new AtlasServiceException(api, e);
                  }
              } else if (clientResponse.getStatus() != ClientResponse.Status.SERVICE_UNAVAILABLE.getStatusCode() &&
                      clientResponse.getStatus() != ClientResponse.Status.INTERNAL_SERVER_ERROR.getStatusCode()) {
                  break;
              } else {
                  LOG.error("attempt #{}: error {} when calling: {}, will retry..", (i + 1), clientResponse.getStatus(), resource);
  
                  sleepBetweenRetries();
              }
  
              i++;
          }
          while (i < getNumberOfRetries());
  
          throw new AtlasServiceException(api, clientResponse);
      }
  
      protected WebResource getResource(API api, String... pathParams) {
          return getResource(service, api, pathParams);
      }
  
      protected WebResource getResource(API api, MultivaluedMap<String, String> queryParams, String... pathParams) {
          WebResource resource = service.path(api.getNormalizedPath());
  
          resource = appendPathParams(resource, pathParams);
          resource = appendQueryParams(queryParams, resource);
  
          return resource;
      }
  
      protected WebResource getResource(API api, MultivaluedMap<String, String> queryParams) {
          return getResource(service, api, queryParams);
      }
  
      protected abstract API formatPathParameters(API api, String... params);
  
      void initializeState(String[] baseUrls, UserGroupInformation ugi, String doAsUser) {
          initializeState(getClientProperties(), baseUrls, ugi, doAsUser);
      }
  
      void initializeState(Configuration configuration, String[] baseUrls, UserGroupInformation ugi, String doAsUser) {
          this.configuration = configuration;
  
          Client client = getClient(configuration, ugi, doAsUser);
  
          if ((!AuthenticationUtil.isKerberosAuthenticationEnabled()) && basicAuthUser != null && basicAuthPassword != null) {
              final HTTPBasicAuthFilter authFilter = new HTTPBasicAuthFilter(basicAuthUser, basicAuthPassword);
  
              client.addFilter(authFilter);
          }
  
          String activeServiceUrl = determineActiveServiceURL(baseUrls, client);
  
          atlasClientContext = new AtlasClientContext(baseUrls, client, ugi, doAsUser);
          service            = client.resource(UriBuilder.fromUri(activeServiceUrl).build());
      }
  
      void sleepBetweenRetries() {
          try {
              Thread.sleep(getSleepBetweenRetriesMs());
          } catch (InterruptedException e) {
              LOG.error("Interrupted from sleeping between retries.", e);
          }
      }
  
      int getNumberOfRetries() {
          return configuration.getInt(AtlasBaseClient.ATLAS_CLIENT_HA_RETRIES_KEY, AtlasBaseClient.DEFAULT_NUM_RETRIES);
      }
  
      boolean isRetryableException(ClientHandlerException che) {
          return che.getCause().getClass().equals(IOException.class) || che.getCause().getClass().equals(ConnectException.class);
      }
  
      void handleClientHandlerException(ClientHandlerException che) {
          if (isRetryableException(che)) {
              atlasClientContext.getClient().destroy();
  
              LOG.warn("Destroyed current context while handling ClientHandlerEception.");
              LOG.warn("Will retry and create new context.");
  
              sleepBetweenRetries();
  
              initializeState(atlasClientContext.getBaseUrls(), atlasClientContext.getUgi(), atlasClientContext.getDoAsUser());
  
              return;
          }
  
          throw che;
      }
  
      @VisibleForTesting
      ObjectNode callAPIWithRetries(API api, Object requestObject, ResourceCreator resourceCreator) throws AtlasServiceException {
          for (int i = 0; i < getNumberOfRetries(); i++) {
              WebResource resource = resourceCreator.createResource();
  
              try {
                  LOG.debug("Using resource {} for {} times", resource.getURI(), i + 1);
  
                  return callAPIWithResource(api, resource, requestObject, ObjectNode.class);
              } catch (ClientHandlerException che) {
                  if (i == (getNumberOfRetries() - 1)) {
                      throw che;
                  }
  
                  LOG.warn("Handled exception in calling api {}", api.getNormalizedPath(), che);
                  LOG.warn("Exception's cause: {}", che.getCause().getClass());
  
                  handleClientHandlerException(che);
              }
          }
  
          throw new AtlasServiceException(api, new RuntimeException("Could not get response after retries."));
      }
  
      @VisibleForTesting
      void setConfiguration(Configuration configuration) {
          this.configuration = configuration;
      }
  
      @VisibleForTesting
      void setService(WebResource resource) {
          this.service = resource;
      }
  
      private AtlasImportResult performImportData(BodyPart requestPart, BodyPart filePart) throws AtlasServiceException {
          MultiPart multipartEntity = new FormDataMultiPart()
                  .bodyPart(requestPart)
                  .bodyPart(filePart);
  
          return callAPI(IMPORT, AtlasImportResult.class, multipartEntity);
      }
  
      private FormDataBodyPart getImportRequestBodyPart(AtlasImportRequest request) {
          return new FormDataBodyPart(IMPORT_REQUEST_PARAMTER, AtlasType.toJson(request), MediaType.APPLICATION_JSON_TYPE);
      }
  
      private String selectActiveServerAddress(Client client, AtlasServerEnsemble serverEnsemble) throws AtlasServiceException {
          List<String> serverInstances     = serverEnsemble.getMembers();
          String       activeServerAddress = null;
  
          for (String serverInstance : serverInstances) {
              LOG.info("Trying with address {}", serverInstance);
  
              activeServerAddress = getAddressIfActive(client, serverInstance);
  
              if (activeServerAddress != null) {
                  LOG.info("Found service {} as active service.", serverInstance);
                  break;
              }
          }
  
          if (activeServerAddress != null) {
              return activeServerAddress;
          } else {
              throw new AtlasServiceException(API_STATUS, new RuntimeException("Could not find any active instance"));
          }
      }
  
      private String getAddressIfActive(Client client, String serverInstance) {
          String activeServerAddress = null;
  
          for (int i = 0; i < getNumberOfRetries(); i++) {
              try {
                  service = client.resource(UriBuilder.fromUri(serverInstance).build());
  
                  String adminStatus = getAdminStatus();
  
                  if (StringUtils.equals(adminStatus, "ACTIVE")) {
                      activeServerAddress = serverInstance;
                      break;
                  } else {
                      LOG.info("attempt #{}: Service {} - is not active. status={}", (i + 1), serverInstance, adminStatus);
                  }
              } catch (Exception e) {
                  LOG.error("attempt #{}: Service {} - could not get status", (i + 1), serverInstance, e);
              }
  
              sleepBetweenRetries();
          }
  
          return activeServerAddress;
      }
  
      private WebResource getResource(WebResource service, String path, String... pathParams) {
          WebResource resource = service.path(path);
  
          resource = appendPathParams(resource, pathParams);
  
          return resource;
      }
  
      private int getSleepBetweenRetriesMs() {
          return configuration.getInt(AtlasBaseClient.ATLAS_CLIENT_HA_SLEEP_INTERVAL_MS_KEY, AtlasBaseClient.DEFAULT_SLEEP_BETWEEN_RETRIES_MS);
      }
  
      // Modify URL to include the path params
      private WebResource getResource(WebResource service, API api, String... pathParams) {
          WebResource resource = service.path(api.getNormalizedPath());
  
          resource = appendPathParams(resource, pathParams);
  
          return resource;
      }
  
      private WebResource getResource(API api, String queryParamKey, List<String> queryParamValues) {
          WebResource resource = service.path(api.getNormalizedPath());
  
          for (String queryParamValue : queryParamValues) {
              if (StringUtils.isNotBlank(queryParamKey) && StringUtils.isNotBlank(queryParamValue)) {
                  resource = resource.queryParam(queryParamKey, queryParamValue);
              }
          }
  
          return resource;
      }
  
      private WebResource appendPathParams(WebResource resource, String[] pathParams) {
          if (pathParams != null) {
              for (String pathParam : pathParams) {
                  resource = resource.path(pathParam);
              }
          }
  
          return resource;
      }
  
      // Modify URL to include the query params
      private WebResource getResource(WebResource service, API api, MultivaluedMap<String, String> queryParams) {
          WebResource resource = service.path(api.getNormalizedPath());
  
          resource = appendQueryParams(queryParams, resource);
  
          return resource;
      }
  
      private WebResource appendQueryParams(MultivaluedMap<String, String> queryParams, WebResource resource) {
          if (null != queryParams && !queryParams.isEmpty()) {
              for (Map.Entry<String, List<String>> entry : queryParams.entrySet()) {
                  for (String value : entry.getValue()) {
                      if (StringUtils.isNotBlank(value)) {
                          resource = resource.queryParam(entry.getKey(), value);
                      }
                  }
              }
          }
  
          return resource;
      }
  
      public static class API {
          private static final Logger LOG = LoggerFactory.getLogger(API.class);
  
          private final String          method;
          private final String          path;
          private final String          consumes;
          private final String          produces;
          private final Response.Status status;
  
          public API(String path, String method, Response.Status status) {
              this(path, method, status, JSON_MEDIA_TYPE, MediaType.APPLICATION_JSON);
          }
  
          public API(String path, String method, Response.Status status, String consumes, String produces) {
              this.path     = path;
              this.method   = method;
              this.status   = status;
              this.consumes = consumes;
              this.produces = produces;
          }
  
          public String getMethod() {
              return method;
          }
  
          public String getPath() {
              return path;
          }
  
          public String getNormalizedPath() {
              // This method used to return Paths.get(path).normalize().toString(), but
              // the use of Paths.get(path) on Windows produces a path with Windows
              // path separators (i.e. back-slashes) which is not valid for a URI
              // and will result in an HTTP 404 status code.
              String resultUri = null;
  
              try {
                  URI uri = new URI(path);
  
                  URI normalizedUri = uri.normalize();
  
                  resultUri = normalizedUri.toString();
              } catch (Exception e) {
                  LOG.error("getNormalizedPath() caught exception for path={}", path, e);
              }
  
              return resultUri;
          }
  
          public Response.Status getExpectedStatus() {
              return status;
          }
  
          public String getConsumes() {
              return consumes;
          }
  
          public String getProduces() {
              return produces;
          }
      }
  
      /**
       * A class to capture input state while creating the client.
       * <p>
       * The information here will be reused when the client is re-initialized on switch-over
       * in case of High Availability.
       */
      private static class AtlasClientContext {
          private final String[]             baseUrls;
          private final Client               client;
          private final String               doAsUser;
          private final UserGroupInformation ugi;
  
          public AtlasClientContext(String[] baseUrls, Client client, UserGroupInformation ugi, String doAsUser) {
              this.baseUrls = baseUrls;
              this.client   = client;
              this.ugi      = ugi;
              this.doAsUser = doAsUser;
          }
  
          public Client getClient() {
              return client;
          }
  
          public String[] getBaseUrls() {
              return baseUrls;
          }
  
          public String getDoAsUser() {
              return doAsUser;
          }
  
          public UserGroupInformation getUgi() {
              return ugi;
          }
      }
  }
  