This is an automated email from the ASF dual-hosted git repository. ramackri pushed a commit to branch RANGER-5656-patch in repository https://gitbox.apache.org/repos/asf/ranger.git
commit dbb20177767f8db379c6ca3b53de167419d225b6 Author: ramk <[email protected]> AuthorDate: Sat Jun 27 16:17:59 2026 +0530 RANGER-5656: TagSync Atlas REST sync fails with AbstractMethodError after RANGER-4076 Jersey 2 migration Replace AtlasClientV2 with HttpURLConnection-based AtlasRESTHttpClient so TagSync does not load Jersey 1.x on the same classpath as Ranger Jersey 2.x REST sink. Remove atlas-client-v1/v2 from tagsync dependencies and assembly tarball. Co-authored-by: Cursor <[email protected]> --- distro/src/main/assembly/tagsync.xml | 3 - tagsync/pom.xml | 54 ------ .../source/atlasrest/AtlasRESTHttpClient.java | 207 +++++++++++++++++++++ .../source/atlasrest/AtlasRESTTagSource.java | 168 +++++++---------- 4 files changed, 277 insertions(+), 155 deletions(-) diff --git a/distro/src/main/assembly/tagsync.xml b/distro/src/main/assembly/tagsync.xml index 6d230c5d1..f919f18d1 100644 --- a/distro/src/main/assembly/tagsync.xml +++ b/distro/src/main/assembly/tagsync.xml @@ -69,9 +69,6 @@ <include>javax.inject:javax.inject</include> <include>javax.xml.bind:jaxb-api</include> <include>joda-time:joda-time</include> - <include>org.apache.atlas:atlas-client-common:jar:${atlas.version}</include> - <include>org.apache.atlas:atlas-client-v1:jar:${atlas.version}</include> - <include>org.apache.atlas:atlas-client-v2:jar:${atlas.version}</include> <include>org.apache.atlas:atlas-common:jar:${atlas.version}</include> <include>org.apache.atlas:atlas-intg:jar:${atlas.version}</include> <include>org.apache.atlas:atlas-notification:jar:${atlas.version}</include> diff --git a/tagsync/pom.xml b/tagsync/pom.xml index 30d5eea5b..5d8ca7a39 100644 --- a/tagsync/pom.xml +++ b/tagsync/pom.xml @@ -155,60 +155,6 @@ <artifactId>jaxb-api</artifactId> <version>${jaxb.api.version}</version> </dependency> - <dependency> - <groupId>org.apache.atlas</groupId> - <artifactId>atlas-client-v1</artifactId> - <version>${atlas.version}</version> - <exclusions> - <exclusion> - <groupId>jakarta.ws.rs</groupId> - <artifactId>jakarta.ws.rs-api</artifactId> - </exclusion> - <exclusion> - <groupId>javax.xml.bind</groupId> - <artifactId>jaxb-api</artifactId> - </exclusion> - <exclusion> - <groupId>javax.xml.bind</groupId> - <artifactId>jaxb-api</artifactId> - </exclusion> - <exclusion> - <groupId>log4j</groupId> - <artifactId>*</artifactId> - </exclusion> - <exclusion> - <groupId>org.slf4j</groupId> - <artifactId>*</artifactId> - </exclusion> - <exclusion> - <groupId>org.springframework</groupId> - <artifactId>*</artifactId> - </exclusion> - </exclusions> - </dependency> - <dependency> - <groupId>org.apache.atlas</groupId> - <artifactId>atlas-client-v2</artifactId> - <version>${atlas.version}</version> - <exclusions> - <exclusion> - <groupId>jakarta.ws.rs</groupId> - <artifactId>jakarta.ws.rs-api</artifactId> - </exclusion> - <exclusion> - <groupId>log4j</groupId> - <artifactId>*</artifactId> - </exclusion> - <exclusion> - <groupId>org.slf4j</groupId> - <artifactId>*</artifactId> - </exclusion> - <exclusion> - <groupId>org.springframework</groupId> - <artifactId>*</artifactId> - </exclusion> - </exclusions> - </dependency> <dependency> <groupId>org.apache.atlas</groupId> <artifactId>atlas-common</artifactId> diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTHttpClient.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTHttpClient.java new file mode 100644 index 000000000..61c91e1ad --- /dev/null +++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTHttpClient.java @@ -0,0 +1,207 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.ranger.tagsync.source.atlasrest; + +import org.apache.atlas.model.discovery.AtlasSearchResult; +import org.apache.atlas.model.discovery.SearchParameters; +import org.apache.atlas.model.typedef.AtlasTypesDef; +import org.apache.atlas.type.AtlasType; +import org.apache.commons.lang3.StringUtils; +import org.apache.hadoop.security.UserGroupInformation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.HttpURLConnection; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.security.PrivilegedExceptionAction; +import java.util.Base64; + +/** + * Minimal Atlas REST client for TagSync. + * + * <p>Uses {@link HttpURLConnection} instead of {@code AtlasClientV2} so TagSync + * does not load the Atlas Jersey 1.x client on the same classpath as Ranger's + * Jersey 2.x REST sink (see RANGER-4076). + */ +public final class AtlasRESTHttpClient { + /** Class logger. */ + private static final Logger LOG = + LoggerFactory.getLogger(AtlasRESTHttpClient.class); + + /** Atlas basic search API path (POST). */ + private static final String SEARCH_PATH = "api/atlas/v2/search/basic"; + + /** Atlas typedef export API path (GET). */ + private static final String TYPEDEFS_PATH = "api/atlas/v2/types/typedefs/"; + + /** Minimum HTTP status treated as an error response. */ + private static final int HTTP_STATUS_CLIENT_ERROR = 400; + + private AtlasRESTHttpClient() { + // utility class + } + + /** + * Search Atlas for entities matching the given parameters. + * + * @param restBaseUrl Atlas REST base URL ending with {@code /} + * @param searchParameters search filter (classification, limit, etc.) + * @param userNamePassword basic-auth credentials; empty for Kerberos + * @param useKerberos {@code true} when TagSync uses a keytab + * @return parsed search result + * @throws IOException when the REST call fails + */ + public static AtlasSearchResult facetedSearch(final String restBaseUrl, + final SearchParameters searchParameters, + final String[] userNamePassword, final boolean useKerberos) + throws IOException { + String body = AtlasType.toJson(searchParameters); + String json = postJson(restBaseUrl + SEARCH_PATH, body, + userNamePassword, useKerberos); + + return AtlasType.fromJson(json, AtlasSearchResult.class); + } + + /** + * Download all Atlas type definitions. + * + * @param restBaseUrl Atlas REST base URL ending with {@code /} + * @param userNamePassword basic-auth credentials; empty for Kerberos + * @param useKerberos {@code true} when TagSync uses a keytab + * @return typedef bundle from Atlas + * @throws IOException when the REST call fails + */ + public static AtlasTypesDef getAllTypeDefs(final String restBaseUrl, + final String[] userNamePassword, final boolean useKerberos) + throws IOException { + String json = getJson(restBaseUrl + TYPEDEFS_PATH, userNamePassword, + useKerberos); + + return AtlasType.fromJson(json, AtlasTypesDef.class); + } + + private static String postJson(final String url, final String body, + final String[] userNamePassword, final boolean useKerberos) + throws IOException { + return execute(url, "POST", body, userNamePassword, useKerberos); + } + + private static String getJson(final String url, + final String[] userNamePassword, final boolean useKerberos) + throws IOException { + return execute(url, "GET", null, userNamePassword, useKerberos); + } + + private static String execute(final String url, final String method, + final String body, final String[] userNamePassword, + final boolean useKerberos) throws IOException { + if (useKerberosAuth(userNamePassword, useKerberos)) { + UserGroupInformation ugi = UserGroupInformation.getLoginUser(); + + ugi.checkTGTAndReloginFromKeytab(); + + try { + return ugi.doAs((PrivilegedExceptionAction<String>) () -> + openAndRead(url, method, body, null, true)); + } catch (InterruptedException interruptedException) { + Thread.currentThread().interrupt(); + throw new IOException("Atlas REST call interrupted", + interruptedException); + } + } + + return openAndRead(url, method, body, userNamePassword, false); + } + + private static boolean useKerberosAuth(final String[] userNamePassword, + final boolean useKerberos) { + return useKerberos && (userNamePassword == null + || userNamePassword.length == 0 + || StringUtils.isBlank(userNamePassword[0])); + } + + private static String openAndRead(final String url, final String method, + final String body, final String[] userNamePassword, + final boolean kerberos) throws IOException { + HttpURLConnection connection = + (HttpURLConnection) new URL(url).openConnection(); + + connection.setRequestMethod(method); + connection.setRequestProperty("Accept", "application/json"); + connection.setRequestProperty("Content-Type", "application/json"); + + if (!kerberos && userNamePassword != null + && userNamePassword.length >= 2 + && StringUtils.isNotEmpty(userNamePassword[0])) { + String credentials = userNamePassword[0] + ":" + + userNamePassword[1]; + String encoded = Base64.getEncoder().encodeToString( + credentials.getBytes(StandardCharsets.UTF_8)); + + connection.setRequestProperty("Authorization", "Basic " + encoded); + } + + if (body != null) { + connection.setDoOutput(true); + byte[] payload = body.getBytes(StandardCharsets.UTF_8); + + connection.setRequestProperty("Content-Length", + String.valueOf(payload.length)); + + try (OutputStream outputStream = connection.getOutputStream()) { + outputStream.write(payload); + } + } + + int status = connection.getResponseCode(); + InputStream stream = status >= HTTP_STATUS_CLIENT_ERROR + ? connection.getErrorStream() + : connection.getInputStream(); + + if (stream == null) { + throw new IOException("Atlas REST " + method + " " + url + + " returned HTTP " + status + " with empty body"); + } + + String response = readStream(stream); + + if (status >= HTTP_STATUS_CLIENT_ERROR) { + throw new IOException("Atlas REST " + method + " " + url + + " failed with HTTP " + status + ": " + response); + } + + if (LOG.isDebugEnabled()) { + LOG.debug("Atlas REST {} {} -> HTTP {}", method, url, status); + } + + return response; + } + + private static String readStream(final InputStream stream) + throws IOException { + byte[] buffer = stream.readAllBytes(); + + return new String(buffer, StandardCharsets.UTF_8); + } +} diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTTagSource.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTTagSource.java index 8ac40cbac..6ae787908 100644 --- a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTTagSource.java +++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTTagSource.java @@ -19,10 +19,6 @@ package org.apache.ranger.tagsync.source.atlasrest; -import org.apache.atlas.AtlasClientV2; -import org.apache.atlas.AtlasServiceException; -import org.apache.atlas.exception.AtlasBaseException; -import org.apache.atlas.model.SearchFilter; import org.apache.atlas.model.TimeBoundary; import org.apache.atlas.model.discovery.AtlasSearchResult; import org.apache.atlas.model.discovery.SearchParameters; @@ -39,7 +35,6 @@ import org.apache.commons.collections.CollectionUtils; import org.apache.commons.collections.MapUtils; import org.apache.commons.lang3.StringUtils; -import org.apache.hadoop.security.UserGroupInformation; import org.apache.ranger.authorization.utils.JsonUtils; import org.apache.ranger.plugin.model.RangerValiditySchedule; import org.apache.ranger.plugin.util.ServiceTags; @@ -233,104 +228,99 @@ private List<RangerAtlasEntityWithTags> getAtlasActiveEntities() { List<RangerAtlasEntityWithTags> ret = new ArrayList<>(); - AtlasClientV2 atlasClient = null; - try { - atlasClient = getAtlasClient(); - } catch (IOException exception) { - LOG.error("Failed to get Atlas client.", exception); + if (restUrls == null || restUrls.length == 0) { + return ret; } - if (atlasClient != null) { - SearchParameters searchParams = new SearchParameters(); - - searchParams.setExcludeDeletedEntities(true); - searchParams.setClassification("*"); - //searchParams.setIncludeSubClassifications(true); - //searchParams.setIncludeSubTypes(true); - searchParams.setIncludeClassificationAttributes(true); - searchParams.setLimit(entitiesBatchSize); - - boolean isMoreData; - int nextStartIndex = 0; - - do { - AtlasTypeRegistry typeRegistry = new AtlasTypeRegistry(); - AtlasTypeRegistry.AtlasTransientTypeRegistry tty = null; - AtlasSearchResult searchResult = null; - boolean commitUpdates = false; - - searchParams.setOffset(nextStartIndex); - isMoreData = false; - - try { - searchResult = atlasClient.facetedSearch(searchParams); - AtlasTypesDef typesDef = atlasClient.getAllTypeDefs(new SearchFilter()); - tty = typeRegistry.lockTypeRegistryForUpdate(); - tty.addTypes(typesDef); - commitUpdates = true; - } catch (AtlasServiceException | AtlasBaseException excp) { - LOG.error("failed to download tags from Atlas", excp); - ret = null; - } catch (Exception unexpectedException) { - LOG.error("Failed to download tags from Atlas due to unexpected exception", unexpectedException); - ret = null; - } finally { - if (tty != null) { - typeRegistry.releaseTypeRegistryForUpdate(tty, commitUpdates); - } + String restBaseUrl = restUrls[0]; + + SearchParameters searchParams = new SearchParameters(); + + searchParams.setExcludeDeletedEntities(true); + searchParams.setClassification("*"); + searchParams.setIncludeClassificationAttributes(true); + searchParams.setLimit(entitiesBatchSize); + + boolean isMoreData; + int nextStartIndex = 0; + + do { + AtlasTypeRegistry typeRegistry = new AtlasTypeRegistry(); + AtlasTypeRegistry.AtlasTransientTypeRegistry tty = null; + AtlasSearchResult searchResult = null; + boolean commitUpdates = false; + + searchParams.setOffset(nextStartIndex); + isMoreData = false; + + try { + searchResult = AtlasRESTHttpClient.facetedSearch(restBaseUrl, + searchParams, userNamePassword, isKerberized); + AtlasTypesDef typesDef = AtlasRESTHttpClient.getAllTypeDefs( + restBaseUrl, userNamePassword, isKerberized); + + tty = typeRegistry.lockTypeRegistryForUpdate(); + tty.addTypes(typesDef); + commitUpdates = true; + } catch (IOException excp) { + LOG.error("failed to download tags from Atlas", excp); + ret = null; + } catch (Exception unexpectedException) { + LOG.error("Failed to download tags from Atlas due to unexpected exception", unexpectedException); + ret = null; + } finally { + if (tty != null) { + typeRegistry.releaseTypeRegistryForUpdate(tty, commitUpdates); } + } - if (commitUpdates && searchResult != null) { - LOG.debug(AtlasType.toJson(searchResult)); + if (commitUpdates && searchResult != null) { + LOG.debug(AtlasType.toJson(searchResult)); - List<AtlasEntityHeader> entityHeaders = searchResult.getEntities(); + List<AtlasEntityHeader> entityHeaders = searchResult.getEntities(); - if (CollectionUtils.isNotEmpty(entityHeaders)) { - nextStartIndex += entityHeaders.size(); - isMoreData = true; + if (CollectionUtils.isNotEmpty(entityHeaders)) { + nextStartIndex += entityHeaders.size(); + isMoreData = true; - for (AtlasEntityHeader header : entityHeaders) { - if (!header.getStatus().equals(AtlasEntity.Status.ACTIVE)) { - LOG.debug("Skipping entity because it is not ACTIVE, header:[{}]", header); - continue; - } + for (AtlasEntityHeader header : entityHeaders) { + if (!header.getStatus().equals(AtlasEntity.Status.ACTIVE)) { + LOG.debug("Skipping entity because it is not ACTIVE, header:[{}]", header); + continue; + } - String typeName = header.getTypeName(); - if (!AtlasResourceMapperUtil.isEntityTypeHandled(typeName)) { - LOG.debug("Not fetching Atlas entities of type:[{}]", typeName); - continue; - } + String typeName = header.getTypeName(); + if (!AtlasResourceMapperUtil.isEntityTypeHandled(typeName)) { + LOG.debug("Not fetching Atlas entities of type:[{}]", typeName); + continue; + } - List<EntityNotificationWrapper.RangerAtlasClassification> allTagsForEntity = new ArrayList<>(); + List<EntityNotificationWrapper.RangerAtlasClassification> allTagsForEntity = new ArrayList<>(); - for (AtlasClassification classification : header.getClassifications()) { - List<EntityNotificationWrapper.RangerAtlasClassification> tags = resolveTag(typeRegistry, classification); - if (tags != null) { - allTagsForEntity.addAll(tags); - } + for (AtlasClassification classification : header.getClassifications()) { + List<EntityNotificationWrapper.RangerAtlasClassification> tags = resolveTag(typeRegistry, classification); + if (tags != null) { + allTagsForEntity.addAll(tags); } + } - if (CollectionUtils.isNotEmpty(allTagsForEntity)) { - RangerAtlasEntity entity = new RangerAtlasEntity(typeName, header.getGuid(), header.getAttributes()); - RangerAtlasEntityWithTags entityWithTags = new RangerAtlasEntityWithTags(entity, allTagsForEntity, typeRegistry); + if (CollectionUtils.isNotEmpty(allTagsForEntity)) { + RangerAtlasEntity entity = new RangerAtlasEntity(typeName, header.getGuid(), header.getAttributes()); + RangerAtlasEntityWithTags entityWithTags = new RangerAtlasEntityWithTags(entity, allTagsForEntity, typeRegistry); - ret.add(entityWithTags); - } + ret.add(entityWithTags); } } } } - while (isMoreData); } + while (isMoreData); - LOG.debug("<== getAtlasActiveEntities()"); + LOG.debug("<== getAtlasActiveEntities(), count={}", ret.size()); return ret; } - /* - * Returns a list of <EntityNotificationWrapper.RangerAtlasClassification> - */ private List<EntityNotificationWrapper.RangerAtlasClassification> resolveTag(AtlasTypeRegistry typeRegistry, AtlasClassification classification) { List<EntityNotificationWrapper.RangerAtlasClassification> ret = new ArrayList<>(); String typeName = classification.getTypeName(); @@ -369,10 +359,8 @@ private List<EntityNotificationWrapper.RangerAtlasClassification> resolveTag(Atl validitySchedules = EntityNotificationWrapper.convertTimeSpecFromAtlasToRanger(validityPeriods); } - // Add most derived classificationType with all attributes ret.add(new EntityNotificationWrapper.RangerAtlasClassification(typeName, allAttributes, validitySchedules)); - // Find base classification types Set<String> superTypeNames = classificationType.getAllSuperTypes(); for (String superTypeName : superTypeNames) { AtlasClassificationType superType = typeRegistry.getClassificationTypeByName(superTypeName); @@ -404,20 +392,4 @@ private List<EntityNotificationWrapper.RangerAtlasClassification> resolveTag(Atl } return ret; } - - private AtlasClientV2 getAtlasClient() throws IOException { - final AtlasClientV2 ret; - - if (isKerberized) { - UserGroupInformation ugi = UserGroupInformation.getLoginUser(); - - ugi.checkTGTAndReloginFromKeytab(); - - ret = new AtlasClientV2(ugi, ugi.getShortUserName(), restUrls); - } else { - ret = new AtlasClientV2(restUrls, userNamePassword); - } - - return ret; - } }
