This is an automated email from the ASF dual-hosted git repository.
ramackri pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ranger.git
The following commit(s) were added to refs/heads/master by this push:
new 5aa6629ba RANGER-5656: TagSync Atlas REST sync fails with
AbstractMethodError after RANGER-4076 Jersey 2 migration (#1035)
5aa6629ba is described below
commit 5aa6629baa3d4bb9e078285c041d3aff2cb22a68
Author: Ramachandran Krishnan <[email protected]>
AuthorDate: Sun Jun 28 07:29:18 2026 +0530
RANGER-5656: TagSync Atlas REST sync fails with AbstractMethodError after
RANGER-4076 Jersey 2 migration (#1035)
* 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.
* RANGER-5656: Address review feedback on AtlasRESTHttpClient and restore
resolveTag comments
Collapse multi-line statements in AtlasRESTHttpClient and restore original
resolveTag block and inline comments in AtlasRESTTagSource.
* RANGER-5656: Use RangerRESTClient for Atlas REST SSL and HA URL failover
Replace HttpURLConnection Atlas client with RangerRESTClient so TagSync
reuses the same SSL and multi-URL failover as TagAdminRESTSink. Initialize
AtlasRESTHttpClient from comma-separated endpoints and keep basic auth when
credentials are configured alongside Kerberos.
* RANGER-5656: Fix checkstyle ImportOrder and apply review style in
AtlasRESTHttpClient
Add blank line between javax and java imports for checkstyle, use single
return in execute(), and keep Atlas REST client initialization on one line.
* Update AtlasRESTHttpClient.java
* Update AtlasRESTTagSource.java
---------
Co-authored-by: ramk <[email protected]>
Co-authored-by: Madhan Neethiraj <[email protected]>
---
distro/src/main/assembly/tagsync.xml | 3 -
tagsync/pom.xml | 54 ------
.../source/atlasrest/AtlasRESTHttpClient.java | 146 ++++++++++++++++
.../source/atlasrest/AtlasRESTTagSource.java | 186 ++++++++++-----------
4 files changed, 233 insertions(+), 156 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..f78f6b0bb
--- /dev/null
+++
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTHttpClient.java
@@ -0,0 +1,146 @@
+/*
+ * 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.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.ranger.plugin.util.RangerRESTClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.ws.rs.core.Response;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+
+/**
+ * Minimal Atlas REST client for TagSync.
+ *
+ * <p>Uses {@link RangerRESTClient} for SSL, HA URL failover, and auth so
TagSync
+ * does not load the Atlas Jersey 1.x client on the same classpath as Ranger's
+ * Jersey 2.x REST sink.
+ */
+public final class AtlasRESTHttpClient {
+ private static final Logger LOG =
LoggerFactory.getLogger(AtlasRESTHttpClient.class);
+
+ private static final String SEARCH_PATH = "/api/atlas/v2/search/basic";
+ private static final String TYPEDEFS_PATH =
"/api/atlas/v2/types/typedefs/";
+
+ private final RangerRESTClient restClient;
+ private final boolean kerberized;
+
+ /**
+ * Create an Atlas REST client backed by {@link RangerRESTClient}.
+ *
+ * @param restUrls comma-separated Atlas REST base URLs
+ * @param sslConfigFile SSL config file for HTTPS endpoints
+ * @param config Hadoop configuration
+ * @param userNamePassword basic-auth credentials when configured
+ * @param useKerberos {@code true} when TagSync uses a keytab
+ */
+ public AtlasRESTHttpClient(final String restUrls, final String
sslConfigFile, final Configuration config, final String[] userNamePassword,
final boolean useKerberos) {
+ this.kerberized = useKerberos;
+ this.restClient = new RangerRESTClient(restUrls, sslConfigFile,
config, "ranger.tagsync");
+
+ if (userNamePassword != null && userNamePassword.length >= 2 &&
StringUtils.isNotEmpty(userNamePassword[0])) {
+ restClient.setBasicAuthInfo(userNamePassword[0],
userNamePassword[1]);
+ }
+
+ restClient.getClient();
+ }
+
+ /**
+ * Search Atlas for entities matching the given parameters.
+ *
+ * @param searchParameters search filter (classification, limit, etc.)
+ * @return parsed search result
+ * @throws Exception when the REST call fails on all configured URLs
+ */
+ public AtlasSearchResult facetedSearch(final SearchParameters
searchParameters) throws Exception {
+ Response response = execute(() -> restClient.post(SEARCH_PATH, null,
searchParameters));
+ String json = readResponseBody(response, SEARCH_PATH);
+
+ return AtlasType.fromJson(json, AtlasSearchResult.class);
+ }
+
+ /**
+ * Download all Atlas type definitions.
+ *
+ * @return typedef bundle from Atlas
+ * @throws Exception when the REST call fails on all configured URLs
+ */
+ public AtlasTypesDef getAllTypeDefs() throws Exception {
+ Response response = execute(() -> restClient.get(TYPEDEFS_PATH, null));
+ String json = readResponseBody(response, TYPEDEFS_PATH);
+
+ return AtlasType.fromJson(json, AtlasTypesDef.class);
+ }
+
+ private Response execute(final RestCall restCall) throws Exception {
+ Response ret;
+
+ if (kerberized) {
+ UserGroupInformation ugi = UserGroupInformation.getLoginUser();
+
+ ugi.checkTGTAndReloginFromKeytab();
+
+ try {
+ ret = ugi.doAs((PrivilegedExceptionAction<Response>)
restCall::run);
+ } catch (InterruptedException interruptedException) {
+ Thread.currentThread().interrupt();
+
+ throw new IOException("Atlas REST call interrupted",
interruptedException);
+ }
+ } else {
+ ret = restCall.run();
+ }
+
+ return ret;
+ }
+
+ private String readResponseBody(final Response response, final String
relativeUrl) throws IOException {
+ if (response == null) {
+ throw new IOException("Atlas REST " + relativeUrl + " failed: no
response from any configured URL");
+ }
+
+ int status = response.getStatus();
+ String body = response.hasEntity() ?
response.readEntity(String.class) : "";
+
+ if (status != Response.Status.OK.getStatusCode()) {
+ throw new IOException("Atlas REST " + relativeUrl + " failed with
HTTP " + status + ": " + body);
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Atlas REST {} -> HTTP {}", relativeUrl, status);
+ }
+
+ return body;
+ }
+
+ @FunctionalInterface
+ private interface RestCall {
+ Response run() throws Exception;
+ }
+}
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..c2547a39f 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;
@@ -77,7 +72,7 @@ public class AtlasRESTTagSource extends AbstractTagSource
implements Runnable {
});
private long sleepTimeBetweenCycleInMillis;
- private String[] restUrls;
+ private AtlasRESTHttpClient atlasRestClient;
private boolean isKerberized;
private String[] userNamePassword;
private int entitiesBatchSize =
TagSyncConfig.DEFAULT_TAGSYNC_ATLASREST_SOURCE_ENTITIES_BATCH_SIZE;
@@ -142,12 +137,11 @@ public boolean initialize(Properties properties) {
LOG.debug("kerberized={}", isKerberized);
if (StringUtils.isNotEmpty(restEndpoint)) {
- this.restUrls = restEndpoint.split(",");
-
- for (int i = 0; i < restUrls.length; i++) {
- if (!restUrls[i].endsWith("/")) {
- restUrls[i] += "/";
- }
+ try {
+ atlasRestClient = new AtlasRESTHttpClient(restEndpoint,
sslConfigFile, TagSyncConfig.getInstance(), userNamePassword, isKerberized);
+ } catch (RuntimeException exception) {
+ LOG.error("Failed to initialize Atlas REST client", exception);
+ ret = false;
}
} else {
LOG.info("AtlasEndpoint not specified, Initial download of
Atlas-entities cannot be done.");
@@ -233,97 +227,107 @@ 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 (atlasRestClient == null) {
+ 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);
- }
+ 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 = atlasRestClient.facetedSearch(searchParams);
+
+ AtlasTypesDef typesDef = atlasRestClient.getAllTypeDefs();
+
+ tty = typeRegistry.lockTypeRegistryForUpdate();
+
+ tty.addTypes(typesDef);
+
+ commitUpdates = true;
+ } catch (IOException excp) {
+ LOG.error("failed to download tags from Atlas", excp);
+
+ ret = null;
+
+ break;
+ } catch (Exception unexpectedException) {
+ LOG.error("Failed to download tags from Atlas due to
unexpected exception", unexpectedException);
+
+ ret = null;
+
+ break;
+ } finally {
+ if (tty != null) {
+ typeRegistry.releaseTypeRegistryForUpdate(tty,
commitUpdates);
}
+ }
- if (commitUpdates && searchResult != null) {
+ if (commitUpdates && searchResult != null) {
+ if (LOG.isDebugEnabled()) {
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);
- String typeName = header.getTypeName();
- if
(!AtlasResourceMapperUtil.isEntityTypeHandled(typeName)) {
- LOG.debug("Not fetching Atlas entities of
type:[{}]", typeName);
- continue;
- }
+ continue;
+ }
-
List<EntityNotificationWrapper.RangerAtlasClassification> allTagsForEntity =
new ArrayList<>();
+ String typeName = header.getTypeName();
- for (AtlasClassification classification :
header.getClassifications()) {
-
List<EntityNotificationWrapper.RangerAtlasClassification> tags =
resolveTag(typeRegistry, classification);
- if (tags != null) {
- allTagsForEntity.addAll(tags);
- }
- }
+ if
(!AtlasResourceMapperUtil.isEntityTypeHandled(typeName)) {
+ LOG.debug("Not fetching Atlas entities of
type:[{}]", typeName);
- if (CollectionUtils.isNotEmpty(allTagsForEntity)) {
- RangerAtlasEntity entity = new
RangerAtlasEntity(typeName, header.getGuid(), header.getAttributes());
- RangerAtlasEntityWithTags entityWithTags = new
RangerAtlasEntityWithTags(entity, allTagsForEntity, typeRegistry);
+ continue;
+ }
+
+
List<EntityNotificationWrapper.RangerAtlasClassification> allTagsForEntity =
new ArrayList<>();
+
+ for (AtlasClassification classification :
header.getClassifications()) {
+
List<EntityNotificationWrapper.RangerAtlasClassification> tags =
resolveTag(typeRegistry, classification);
- ret.add(entityWithTags);
+ 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);
+
+ ret.add(entityWithTags);
+ }
}
}
}
- while (isMoreData);
}
+ while (isMoreData);
- LOG.debug("<== getAtlasActiveEntities()");
+ LOG.debug("<== getAtlasActiveEntities(), count={}", ret == null ? 0 :
ret.size());
return ret;
}
@@ -404,20 +408,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;
- }
}
