You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ranger.apache.org by ra...@apache.org on 2026/07/04 15:59:51 UTC
(ranger) branch ranger-2.9 updated: RANGER-5658: Tag Sync: remove atlas.kafka.zookeeper.connect and fix Atlas Kafka consumer for Kafka 3.9 (#1045)
This is an automated email from the ASF dual-hosted git repository.
ramackri pushed a commit to branch ranger-2.9
in repository https://gitbox.apache.org/repos/asf/ranger.git
The following commit(s) were added to refs/heads/ranger-2.9 by this push:
new f7950da63 RANGER-5658: Tag Sync: remove atlas.kafka.zookeeper.connect and fix Atlas Kafka consumer for Kafka 3.9 (#1045)
f7950da63 is described below
commit f7950da637f951078f8c41d5172190e6a2a63aa0
Author: Ramachandran Krishnan <ra...@gmail.com>
AuthorDate: Sat Jul 4 21:29:46 2026 +0530
RANGER-5658: Tag Sync: remove atlas.kafka.zookeeper.connect and fix Atlas Kafka consumer for Kafka 3.9 (#1045)
Backport of deaf13df3 (#1037) to ranger-2.9.
- Remove obsolete atlas.kafka.zookeeper.connect requirement from Tag Sync
- Fix Atlas Kafka consumer for Kafka 3.9 / KRaft (PLAINTEXT and Kerberos)
- Load atlas-application.properties via ApplicationProperties and -Datlas.conf
- Fix TagSync setup.py rangerTagsync_password credential for jceks
- Add AtlasTagSourceConfigTest and operator documentation
Co-authored-by: ramk <ra...@cloudera.com>
---
dev-support/README-TAGSYNC-ATLAS-KAFKA-CONFIG.md | 202 +++++++
.../tagsync/ranger-tagsync-install.properties | 12 +-
tagsync/conf/templates/installprop2xml.properties | 1 -
tagsync/scripts/install.properties | 3 +-
tagsync/scripts/ranger-tagsync-services.sh | 3 +-
tagsync/scripts/setup.py | 28 +-
.../tagsync/source/atlas/AtlasTagSource.java | 588 ++++++++++-----------
.../source/atlas/AtlasTagSourceConfigTest.java | 57 ++
8 files changed, 577 insertions(+), 317 deletions(-)
diff --git a/dev-support/README-TAGSYNC-ATLAS-KAFKA-CONFIG.md b/dev-support/README-TAGSYNC-ATLAS-KAFKA-CONFIG.md
new file mode 100644
index 000000000..1f6c77c42
--- /dev/null
+++ b/dev-support/README-TAGSYNC-ATLAS-KAFKA-CONFIG.md
@@ -0,0 +1,202 @@
+# Tag Sync — Atlas Kafka configuration (Kafka 3.9+ / KRaft)
+
+Reference for operators and developers configuring Ranger Tag Sync with the **Atlas Kafka** tag source (`ranger.tagsync.source.atlas`).
+
+Related Jira: [RANGER-5658](https://issues.apache.org/jira/browse/RANGER-5658) — remove obsolete `atlas.kafka.zookeeper.connect` and fix Atlas Kafka consumer setup for PLAINTEXT and Kerberos brokers.
+
+---
+
+## Summary
+
+Tag Sync consumes Atlas entity notifications on the **`ATLAS_ENTITIES`** topic using Atlas's `KafkaNotification` / standard **`kafka-clients`** (`bootstrap.servers`, consumer group, security protocol). It does **not** connect to Kafka via ZooKeeper.
+
+The property `atlas.kafka.zookeeper.connect` was legacy configuration. Ranger previously **required** it at startup even though Atlas 2.4 `KafkaNotification` never uses it for consumer creation, and Kafka 3.x / KRaft clients use `bootstrap.servers` only. That requirement has been **removed**.
+
+**Both PLAINTEXT and Kerberos (SASL) Atlas Kafka are supported.** Set `TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL` to match how Atlas publishes entity notifications.
+
+---
+
+## Supported Atlas Kafka security modes
+
+| Mode | `TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL` | Extra install properties | `setup.py` writes Kerberos JAAS? |
+|------|---------------------------------------------|--------------------------|----------------------------------|
+| **PLAINTEXT** | `PLAINTEXT` | Bootstrap + consumer group only | **No** |
+| **Kerberos** | `SASL_PLAINTEXT` or `SASL_SSL` | `TAG_SOURCE_ATLAS_KAFKA_SERVICE_NAME`, `TAG_SOURCE_ATLAS_KERBEROS_PRINCIPAL`, `TAG_SOURCE_ATLAS_KERBEROS_KEYTAB` | **Yes** (when `is_secure=true`) |
+
+`setup.py` uses `atlas_kafka_uses_kerberos()` — Kerberos JAAS for the Atlas Kafka client is emitted **only** when the security protocol starts with `SASL`. This allows Ranger to run with Kerberos (`is_secure=true`) while consuming from a **PLAINTEXT** Atlas metadata bus (common in Atlas+Ranger coexist / dev stacks).
+
+**Important:** Atlas Kafka security is independent of how Tag Sync authenticates to **Ranger Admin** for tag upload (basic auth or Ranger Kerberos via `core-site.xml` / jceks). Configure each path separately.
+
+---
+
+## Required configuration (Atlas Kafka source)
+
+Set these in `install.properties` (installer writes `conf/atlas-application.properties`):
+
+| Install property | Atlas property | Required |
+|------------------|----------------|----------|
+| `TAG_SOURCE_ATLAS_ENABLED` | (enables `ranger.tagsync.source.atlas` in site.xml) | Yes, for Kafka source |
+| `TAG_SOURCE_ATLAS_KAFKA_BOOTSTRAP_SERVERS` | `atlas.kafka.bootstrap.servers` | Yes |
+| `TAG_SOURCE_ATLAS_KAFKA_ENTITIES_GROUP_ID` | `atlas.kafka.entities.group.id` | Yes |
+| `TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL` | `atlas.kafka.security.protocol` | Yes (defaults to PLAINTEXT if omitted in logic) |
+| `TAG_SOURCE_ATLAS_KAFKA_SERVICE_NAME` | `atlas.kafka.sasl.kerberos.service.name` | If Kerberos |
+| `TAG_SOURCE_ATLAS_KERBEROS_PRINCIPAL` / `KEYTAB` | `atlas.jaas.KafkaClient.*` | If Kerberos |
+
+### Example — PLAINTEXT (Atlas metadata bus / KRaft coexist)
+
+Use when Atlas publishes `ATLAS_ENTITIES` on a **separate PLAINTEXT** metadata Kafka (e.g. `atlas-kafka` in Atlas+Ranger coexist stacks).
+
+`install.properties`:
+
+```properties
+TAG_SOURCE_ATLAS_ENABLED=true
+TAG_SOURCE_ATLAS_KAFKA_BOOTSTRAP_SERVERS = atlas-kafka.example.com:9092
+TAG_SOURCE_ATLAS_KAFKA_ENTITIES_GROUP_ID = ranger_entities_consumer
+TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL = PLAINTEXT
+TAG_SOURCE_ATLASREST_ENABLED=false
+TAG_SOURCE_FILE_ENABLED=false
+```
+
+Generated `conf/atlas-application.properties`:
+
+```properties
+atlas.kafka.bootstrap.servers=atlas-kafka.example.com:9092
+atlas.kafka.entities.group.id=ranger_entities_consumer
+atlas.kafka.security.protocol=PLAINTEXT
+```
+
+### Example — Kerberos (secured Atlas Kafka)
+
+Use when Atlas entity notifications are on a **SASL** Kafka broker. In **ranger-docker**, that is typically `ranger-kafka.rangernw:9092` (not PLAINTEXT).
+
+`install.properties`:
+
+```properties
+TAG_SOURCE_ATLAS_ENABLED=true
+TAG_SOURCE_ATLAS_KAFKA_BOOTSTRAP_SERVERS = ranger-kafka.rangernw:9092
+TAG_SOURCE_ATLAS_KAFKA_ENTITIES_GROUP_ID = ranger_entities_consumer
+TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL = SASL_PLAINTEXT
+TAG_SOURCE_ATLAS_KAFKA_SERVICE_NAME = kafka
+TAG_SOURCE_ATLAS_KERBEROS_PRINCIPAL = rangertagsync/ranger-tagsync.rangernw@EXAMPLE.COM
+TAG_SOURCE_ATLAS_KERBEROS_KEYTAB = /etc/keytabs/rangertagsync.keytab
+```
+
+Requires `is_secure=true` in install so Kerberos JAAS is written for the Atlas Kafka client.
+
+See also: `dev-support/ranger-docker/scripts/tagsync/ranger-tagsync-install.properties`.
+
+---
+
+## Starting Tag Sync
+
+### Native / tarball install
+
+After `setup.sh` in the tagsync package:
+
+```bash
+cd ${RANGER_TAGSYNC_HOME}/tagsync
+./ranger-tagsync-services.sh start # stop | restart
+```
+
+Or via the init wrapper (runs as `ranger` user):
+
+```bash
+/usr/bin/ranger-tagsync.sh start # stop | restart | status
+```
+
+### JVM flag (Atlas Kafka source)
+
+`ranger-tagsync-services.sh` must pass the install conf directory as Atlas config root:
+
+```bash
+-Datlas.conf="${TAGSYNC_CONF_DIR}"
+```
+
+Without this, `AtlasTagSource` may not load installer-generated `atlas-application.properties` and Kafka consumer startup will fail or use wrong settings. Custom systemd/k8s wrappers must include the same flag.
+
+### Docker (ranger-docker)
+
+Default sample (`ranger-tagsync-install.properties`): **file source on** (no Atlas server in ranger-docker), **Atlas REST off**, **Atlas Kafka off**. For Atlas+Ranger coexist stacks, set `TAG_SOURCE_ATLAS_ENABLED=true`, `TAG_SOURCE_ATLASREST_ENABLED=false`, and `TAG_SOURCE_FILE_ENABLED=false` to consume `ATLAS_ENTITIES` from Kafka.
+
+```bash
+cd dev-support/ranger-docker
+docker compose -f docker-compose.ranger.yml -f docker-compose.ranger-tagsync.yml up -d
+```
+
+The container entrypoint runs `setup.sh` then `ranger-tagsync-services.sh start`.
+
+After changing `install.properties` or Atlas Kafka settings, force re-setup:
+
+```bash
+docker exec ranger-tagsync rm -f /opt/ranger/ranger-*-tagsync/.setupDone
+docker restart ranger-tagsync
+```
+
+### Verify Tag Sync is running
+
+```bash
+# Process
+ps -ef | grep TagSynchronizer
+
+# Atlas Kafka config on disk
+grep -E 'bootstrap.servers|security.protocol|zookeeper' \
+ /etc/ranger/tagsync/conf/atlas-application.properties
+
+# Logs
+tail -f /var/log/ranger/tagsync/tagsync-*.log
+```
+
+Expect: bootstrap servers and consumer group set; **no** `atlas.kafka.zookeeper.connect`; for PLAINTEXT, **no** `atlas.jaas.KafkaClient` entries.
+
+---
+
+## Removed configuration
+
+| Property | Status |
+|----------|--------|
+| `atlas.kafka.zookeeper.connect` | **No longer required** — not generated by `setup.py`, not validated by `AtlasTagSource` |
+
+Existing deployments may still have this key in `atlas-application.properties`; it is harmless and can be removed on the next config refresh.
+
+---
+
+## ZooKeeper still used elsewhere (do not confuse)
+
+| Config | Purpose | Needed for KRaft Kafka? |
+|--------|---------|-------------------------|
+| `atlas.kafka.zookeeper.connect` | Legacy Atlas/Kafka config | **No** (removed) |
+| `ranger-tagsync.server.ha.zookeeper.connect` | Tag Sync active/passive leader election | **Only when HA enabled** |
+
+---
+
+## Alternative: Atlas REST source
+
+If Tag Sync reads tags over Atlas REST instead of Kafka, Atlas Kafka properties do not apply:
+
+```properties
+TAG_SOURCE_ATLASREST_ENABLED=true
+TAG_SOURCE_ATLASREST_ENDPOINT = http://localhost:21000
+TAG_SOURCE_ATLAS_ENABLED=false
+```
+
+Enable via `ranger.tagsync.source.atlasrest` in `ranger-tagsync-site.xml`.
+
+---
+
+## Test plan
+
+- [ ] Fresh install: `atlas-application.properties` has no `zookeeper.connect`
+- [ ] PLAINTEXT: `is_secure=true` + `security.protocol=PLAINTEXT` → no Kerberos JAAS in atlas props
+- [ ] SASL: Kerberos JAAS written when `security.protocol=SASL_PLAINTEXT`
+- [ ] Tag Sync starts with `-Datlas.conf` and consumes `ATLAS_ENTITIES` from Kafka 3.9.x (KRaft)
+- [ ] Kerberos Atlas Kafka: consumer connects with principal/keytab
+- [ ] Tag Sync HA smoke test (if `ranger-tagsync.server.ha.enabled=true`)
+
+---
+
+## References
+
+- Tag Sync Atlas source: `tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasTagSource.java`
+- Atlas notification: `org.apache.atlas.kafka.KafkaNotification` in `atlas-notification`
+- Docker sample: `dev-support/ranger-docker/scripts/tagsync/ranger-tagsync-install.properties`
+- Ranger docker overview: `dev-support/ranger-docker/README.md`
diff --git a/dev-support/ranger-docker/scripts/tagsync/ranger-tagsync-install.properties b/dev-support/ranger-docker/scripts/tagsync/ranger-tagsync-install.properties
index 7f0c5ecba..14101965d 100755
--- a/dev-support/ranger-docker/scripts/tagsync/ranger-tagsync-install.properties
+++ b/dev-support/ranger-docker/scripts/tagsync/ranger-tagsync-install.properties
@@ -25,21 +25,20 @@ TAG_DEST_RANGER_ENDPOINT = http://ranger:6080
# SSL config file name for HTTPS messages to tag destination - Ranger
TAG_DEST_RANGER_SSL_CONFIG_FILENAME =
+# Atlas Kafka source (bootstrap.servers only; no ZooKeeper). Off by default in ranger-docker.
+# Enable for Atlas+Ranger coexist stacks that publish ATLAS_ENTITIES (see README-TAGSYNC-ATLAS-KAFKA-CONFIG.md).
TAG_SOURCE_ATLAS_ENABLED = false
-# Endpoint specifications needed by Atlas
TAG_SOURCE_ATLAS_KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR = 1
TAG_SOURCE_ATLAS_KAFKA_BOOTSTRAP_SERVERS = ranger-kafka.rangernw:9092
-TAG_SOURCE_ATLAS_KAFKA_ZOOKEEPER_CONNECT = ranger-zk.rangernw:2181
TAG_SOURCE_ATLAS_KAFKA_ENTITIES_GROUP_ID = ranger_entities_consumer
-
-TAG_SOURCE_ATLAS_KAFKA_SERVICE_NAME = kafka
TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL = SASL_PLAINTEXT
+TAG_SOURCE_ATLAS_KAFKA_SERVICE_NAME = kafka
TAG_SOURCE_ATLAS_KERBEROS_PRINCIPAL = rangertagsync/ranger-tagsync.rangernw@EXAMPLE.COM
TAG_SOURCE_ATLAS_KERBEROS_KEYTAB = /etc/keytabs/rangertagsync.keytab
+# No Atlas server in ranger-docker — use file source by default (not Atlas REST).
TAG_SOURCE_ATLASREST_ENABLED = false
-
TAG_SOURCE_ATLASREST_ENDPOINT = http://localhost:21000
TAG_SOURCE_ATLASREST_DOWNLOAD_INTERVAL_IN_MILLIS = 900000
@@ -68,7 +67,8 @@ TAG_SOURCE_FILE_CHECK_INTERVAL_IN_MILLIS = 60000
# TAGSYNC_ATLAS_TO_RANGER_SERVICE_MAPPING = clusterName,hive,rangerServiceName
#
-TAGSYNC_ATLAS_TO_RANGER_SERVICE_MAPPING=
+# Atlas Hive entities use @primary suffix → Ranger service dev_hive
+TAGSYNC_ATLAS_TO_RANGER_SERVICE_MAPPING = primary,hive,dev_hive
# A comma separated list of custom mapper class names which convert Atlas entities to
# RangerServiceResource structures are specified here. If there are no custom mappers,
diff --git a/tagsync/conf/templates/installprop2xml.properties b/tagsync/conf/templates/installprop2xml.properties
index d6f289a55..aea6537ee 100644
--- a/tagsync/conf/templates/installprop2xml.properties
+++ b/tagsync/conf/templates/installprop2xml.properties
@@ -24,7 +24,6 @@ TAG_SOURCE_ATLAS_ENABLED = ranger.tagsync.source.atlas
TAG_SOURCE_ATLAS_KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR = atlas.kafka.offsets.topic.replication.factor
TAG_SOURCE_ATLAS_KAFKA_BOOTSTRAP_SERVERS = atlas.kafka.bootstrap.servers
-TAG_SOURCE_ATLAS_KAFKA_ZOOKEEPER_CONNECT = atlas.kafka.zookeeper.connect
TAG_SOURCE_ATLAS_KAFKA_ENTITIES_GROUP_ID = atlas.kafka.entities.group.id
TAG_SOURCE_ATLAS_KAFKA_SERVICE_NAME = atlas.kafka.sasl.kerberos.service.name
diff --git a/tagsync/scripts/install.properties b/tagsync/scripts/install.properties
index ae2e555eb..278df0ff8 100644
--- a/tagsync/scripts/install.properties
+++ b/tagsync/scripts/install.properties
@@ -27,9 +27,8 @@ TAG_DEST_RANGER_SSL_CONFIG_FILENAME =
TAG_SOURCE_ATLAS_ENABLED = true
-# Endpoint specifications needed by Atlas
+# Kafka settings for Atlas notification consumer (bootstrap.servers only; no ZooKeeper)
TAG_SOURCE_ATLAS_KAFKA_BOOTSTRAP_SERVERS = localhost:6667
-TAG_SOURCE_ATLAS_KAFKA_ZOOKEEPER_CONNECT = localhost:2181
TAG_SOURCE_ATLAS_KAFKA_ENTITIES_GROUP_ID = ranger_entities_consumer
TAG_SOURCE_ATLAS_KAFKA_SERVICE_NAME = kafka
diff --git a/tagsync/scripts/ranger-tagsync-services.sh b/tagsync/scripts/ranger-tagsync-services.sh
index ab23ab9b2..ef75644ec 100755
--- a/tagsync/scripts/ranger-tagsync-services.sh
+++ b/tagsync/scripts/ranger-tagsync-services.sh
@@ -95,9 +95,10 @@ if [ "${action}" == "START" ]; then
if [ -z "${TAGSYNC_CONF_DIR}" ]; then
TAGSYNC_CONF_DIR=${cdir}/conf
fi
+ export TAGSYNC_CONF_DIR
SLEEP_TIME_AFTER_START=5
- nohup java -Dproc_rangertagsync ${JAVA_OPTS} -Dlogdir="${RANGER_TAGSYNC_LOG_DIR}" -Dlogback.configurationFile=file:${TAGSYNC_CONF_DIR}/logback.xml -Duser=${USER} -Dhostname=${HOSTNAME} -cp "${cp}" org.apache.ranger.tagsync.process.TagSynchronizer > ${RANGER_TAGSYNC_LOG_DIR}/tagsync.out 2>&1 &
+ nohup java -Dproc_rangertagsync ${JAVA_OPTS} -Datlas.conf="${TAGSYNC_CONF_DIR}" -Dlogdir="${RANGER_TAGSYNC_LOG_DIR}" -Dlogback.configurationFile=file:${TAGSYNC_CONF_DIR}/logback.xml -Duser=${USER} -Dhostname=${HOSTNAME} -cp "${cp}" org.apache.ranger.tagsync.process.TagSynchronizer > ${RANGER_TAGSYNC_LOG_DIR}/tagsync.out 2>&1 &
VALUE_OF_PID=$!
echo "Starting Apache Ranger Tagsync Service"
sleep $SLEEP_TIME_AFTER_START
diff --git a/tagsync/scripts/setup.py b/tagsync/scripts/setup.py
index 1fbf5f181..c7ca4b3b0 100755
--- a/tagsync/scripts/setup.py
+++ b/tagsync/scripts/setup.py
@@ -68,7 +68,6 @@
initPrefixList = ['S99', 'K00']
TAGSYNC_ATLAS_KAFKA_ENDPOINTS_KEY = 'TAG_SOURCE_ATLAS_KAFKA_BOOTSTRAP_SERVERS'
-TAGSYNC_ATLAS_ZOOKEEPER_ENDPOINT_KEY = 'TAG_SOURCE_ATLAS_KAFKA_ZOOKEEPER_CONNECT'
TAGSYNC_ATLAS_KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR_KEY = 'TAG_SOURCE_ATLAS_KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR'
TAGSYNC_ATLAS_CONSUMER_GROUP_KEY = 'TAG_SOURCE_ATLAS_KAFKA_ENTITIES_GROUP_ID'
@@ -216,6 +215,15 @@ def updatePropertyInJCKSFile(jcksFileName,propName,value):
sys.exit(1)
return ret
+def atlas_kafka_uses_kerberos(props):
+ """Kerberos JAAS for Atlas Kafka consumer only when SASL is configured (not PLAINTEXT)."""
+ if not configure_security:
+ return False
+ protocol = props.get(TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL_KEY, 'PLAINTEXT')
+ if protocol is None:
+ return False
+ return protocol.strip().upper().startswith('SASL')
+
def convertInstallPropsToXML(props):
directKeyMap = getPropertiesConfigMap(join(installTemplateDirName,install2xmlMapFileName))
ret = {}
@@ -225,32 +233,31 @@ def convertInstallPropsToXML(props):
atlas_principal = ''
atlas_keytab = ''
+ atlas_kafka_kerberos = atlas_kafka_uses_kerberos(props)
for k,v in props.items():
if (k in list(directKeyMap)):
newKey = directKeyMap[k]
if (k == TAGSYNC_ATLAS_KAFKA_ENDPOINTS_KEY):
atlasOutFile.write(newKey + "=" + v + "\n")
- elif (k == TAGSYNC_ATLAS_ZOOKEEPER_ENDPOINT_KEY):
- atlasOutFile.write(newKey + "=" + v + "\n")
elif (k == TAGSYNC_ATLAS_KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR_KEY):
atlasOutFile.write(newKey + "=" + v + "\n")
elif (k == TAGSYNC_ATLAS_CONSUMER_GROUP_KEY):
atlasOutFile.write(newKey + "=" + v + "\n")
- elif (configure_security and k == TAG_SOURCE_ATLAS_KAKFA_SERVICE_NAME_KEY):
+ elif (k == TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL_KEY):
atlasOutFile.write(newKey + "=" + v + "\n")
- elif (configure_security and k == TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL_KEY):
+ elif (atlas_kafka_kerberos and k == TAG_SOURCE_ATLAS_KAKFA_SERVICE_NAME_KEY):
atlasOutFile.write(newKey + "=" + v + "\n")
- elif (configure_security and k == TAG_SOURCE_ATLAS_KERBEROS_PRINCIPAL_KEY):
+ elif (atlas_kafka_kerberos and k == TAG_SOURCE_ATLAS_KERBEROS_PRINCIPAL_KEY):
atlas_principal = v
- elif (configure_security and k == TAG_SOURCE_ATLAS_KERBEROS_KEYTAB_KEY):
+ elif (atlas_kafka_kerberos and k == TAG_SOURCE_ATLAS_KERBEROS_KEYTAB_KEY):
atlas_keytab = v
else:
ret[newKey] = v
else:
print("INFO: Direct Key not found:%s" % (k))
- if (configure_security):
+ if atlas_kafka_kerberos and atlas_principal and atlas_keytab:
atlasOutFile.write("atlas.jaas.KafkaClient.loginModuleName = com.sun.security.auth.module.Krb5LoginModule" + "\n")
atlasOutFile.write("atlas.jaas.KafkaClient.loginModuleControlFlag = required" + "\n")
atlasOutFile.write("atlas.jaas.KafkaClient.option.useKeyTab = true" + "\n")
@@ -443,7 +450,10 @@ def main():
mergeProps['ranger.tagsync.dest.ranger.username'] = 'rangertagsync'
if (tagsyncKSPath != ''):
- tagadminPasswd = 'rangertagsync'
+ if ('rangerTagsync_password' in globalDict and globalDict['rangerTagsync_password'] != ''):
+ tagadminPasswd = globalDict['rangerTagsync_password']
+ else:
+ tagadminPasswd = 'rangertagsync'
tagadminAlias = 'tagadmin.user.password'
updatePropertyInJCKSFile(tagsyncKSPath,tagadminAlias,tagadminPasswd)
os.chown(tagsyncKSPath,ownerId,groupId)
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasTagSource.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasTagSource.java
index 7688c1d90..95019a98f 100644
--- a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasTagSource.java
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasTagSource.java
@@ -19,25 +19,26 @@
package org.apache.ranger.tagsync.source.atlas;
+import org.apache.atlas.ApplicationProperties;
+import org.apache.atlas.AtlasException;
+import org.apache.atlas.kafka.AtlasKafkaMessage;
import org.apache.atlas.kafka.NotificationProvider;
import org.apache.atlas.model.notification.EntityNotification;
import org.apache.atlas.notification.NotificationConsumer;
import org.apache.atlas.notification.NotificationInterface;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
+import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.common.TopicPartition;
import org.apache.ranger.authorization.utils.JsonUtils;
import org.apache.ranger.plugin.util.ServiceTags;
import org.apache.ranger.tagsync.model.AbstractTagSource;
-import org.apache.atlas.kafka.AtlasKafkaMessage;
-import org.apache.kafka.common.TopicPartition;
import org.apache.ranger.tagsync.process.TagSyncConfig;
import org.apache.ranger.tagsync.source.atlasrest.RangerAtlasEntityWithTags;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -45,299 +46,290 @@
import java.util.Properties;
public class AtlasTagSource extends AbstractTagSource {
- private static final Logger LOG = LoggerFactory.getLogger(AtlasTagSource.class);
-
- public static final String TAGSYNC_ATLAS_PROPERTIES_FILE_NAME = "atlas-application.properties";
-
- public static final String TAGSYNC_ATLAS_KAFKA_ENDPOINTS = "atlas.kafka.bootstrap.servers";
- public static final String TAGSYNC_ATLAS_ZOOKEEPER_ENDPOINT = "atlas.kafka.zookeeper.connect";
- public static final String TAGSYNC_ATLAS_CONSUMER_GROUP = "atlas.kafka.entities.group.id";
-
- public static final int MAX_WAIT_TIME_IN_MILLIS = 1000;
-
- private int maxBatchSize;
-
- private ConsumerRunnable consumerTask;
- private Thread myThread = null;
-
- @Override
- public boolean initialize(Properties properties) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("==> AtlasTagSource.initialize()");
- }
-
- Properties atlasProperties = new Properties();
-
- boolean ret = AtlasResourceMapperUtil.initializeAtlasResourceMappers(properties);
-
- if (ret) {
-
- InputStream inputStream = getClass().getClassLoader().getResourceAsStream(TAGSYNC_ATLAS_PROPERTIES_FILE_NAME);
-
- if (inputStream != null) {
- try {
- atlasProperties.load(inputStream);
- } catch (Exception exception) {
- ret = false;
- LOG.error("Cannot load Atlas application properties file, file-name:" + TAGSYNC_ATLAS_PROPERTIES_FILE_NAME, exception);
- } finally {
- try {
- inputStream.close();
- } catch (IOException ioException) {
- LOG.error("Cannot close Atlas application properties file, file-name:" + TAGSYNC_ATLAS_PROPERTIES_FILE_NAME, ioException);
- }
- }
- } else {
- ret = false;
- LOG.error("Cannot find Atlas application properties file");
- }
- }
-
- if (ret) {
- if (StringUtils.isBlank(atlasProperties.getProperty(TAGSYNC_ATLAS_KAFKA_ENDPOINTS))) {
- ret = false;
- LOG.error("Value of property '" + TAGSYNC_ATLAS_KAFKA_ENDPOINTS + "' is not specified!");
- }
- if (StringUtils.isBlank(atlasProperties.getProperty(TAGSYNC_ATLAS_ZOOKEEPER_ENDPOINT))) {
- ret = false;
- LOG.error("Value of property '" + TAGSYNC_ATLAS_ZOOKEEPER_ENDPOINT + "' is not specified!");
- }
- if (StringUtils.isBlank(atlasProperties.getProperty(TAGSYNC_ATLAS_CONSUMER_GROUP))) {
- ret = false;
- LOG.error("Value of property '" + TAGSYNC_ATLAS_CONSUMER_GROUP + "' is not specified!");
- }
- }
-
- if (ret) {
- NotificationInterface notification = NotificationProvider.get();
- List<NotificationConsumer<EntityNotification>> iterators = notification.createConsumers(NotificationInterface.NotificationType.ENTITIES, 1);
-
- consumerTask = new ConsumerRunnable(iterators.get(0));
- }
-
- maxBatchSize = TagSyncConfig.getSinkMaxBatchSize(properties);
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("<== AtlasTagSource.initialize(), result=" + ret);
- }
- return ret;
- }
-
- @Override
- public boolean start() {
- if (LOG.isDebugEnabled()) {
- LOG.debug("==> AtlasTagSource.start()");
- }
- if (consumerTask == null) {
- LOG.error("No consumerTask!!!");
- } else {
- myThread = new Thread(consumerTask);
- myThread.setDaemon(true);
- myThread.start();
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug("<== AtlasTagSource.start()");
- }
- return myThread != null;
- }
-
- @Override
- public void stop() {
- if (myThread != null && myThread.isAlive()) {
- myThread.interrupt();
- }
- }
-
- private static String getPrintableEntityNotification(EntityNotificationWrapper notification) {
- StringBuilder sb = new StringBuilder();
-
- sb.append("{ Notification-Type: ").append(notification.getOpType()).append(", ");
+ private static final Logger LOG = LoggerFactory.getLogger(AtlasTagSource.class);
+
+ public static final String TAGSYNC_ATLAS_PROPERTIES_FILE_NAME = "atlas-application.properties";
+ public static final String TAGSYNC_ATLAS_KAFKA_ENDPOINTS = "atlas.kafka.bootstrap.servers";
+ public static final String TAGSYNC_ATLAS_CONSUMER_GROUP = "atlas.kafka.entities.group.id";
+ public static final int MAX_WAIT_TIME_IN_MILLIS = 1000;
+
+ private int maxBatchSize;
+ private ConsumerRunnable consumerTask;
+ private Thread myThread;
+
+ @Override
+ public boolean initialize(Properties properties) {
+ LOG.debug("==> AtlasTagSource.initialize()");
+
+ boolean ret = AtlasResourceMapperUtil.initializeAtlasResourceMappers(properties);
+
+ if (ret) {
+ Configuration atlasConfiguration = loadAtlasConfiguration();
+
+ ret = validateRequiredAtlasKafkaProperties(atlasConfiguration);
+
+ if (ret) {
+ NotificationInterface notification = NotificationProvider.get();
+ List<NotificationConsumer<EntityNotification>> iterators = notification.createConsumers(NotificationInterface.NotificationType.ENTITIES, 1);
+
+ maxBatchSize = TagSyncConfig.getSinkMaxBatchSize(properties);
+ consumerTask = new ConsumerRunnable(iterators.get(0));
+ }
+ }
+
+ LOG.debug("<== AtlasTagSource.initialize(), result={}", ret);
+
+ return ret;
+ }
+
+ @Override
+ public boolean start() {
+ LOG.debug("==> AtlasTagSource.start()");
+
+ boolean ret = false;
+
+ if (consumerTask == null) {
+ LOG.error("No consumerTask!!!");
+ } else {
+ myThread = new Thread(consumerTask);
+
+ myThread.setDaemon(true);
+ myThread.start();
+
+ ret = true;
+ }
+
+ LOG.debug("<== AtlasTagSource.start(): ret={}", ret);
+
+ return ret;
+ }
+
+ @Override
+ public void stop() {
+ if (myThread != null && myThread.isAlive()) {
+ myThread.interrupt();
+ }
+ }
+
+ boolean validateRequiredAtlasKafkaProperties(Properties atlasProperties) {
+ boolean ret = true;
+
+ if (atlasProperties == null) {
+ ret = false;
+ } else {
+ if (StringUtils.isBlank(atlasProperties.getProperty(TAGSYNC_ATLAS_KAFKA_ENDPOINTS))) {
+ LOG.error("missing value for mandatory property '{}'", TAGSYNC_ATLAS_KAFKA_ENDPOINTS);
+
+ ret = false;
+ }
+
+ if (StringUtils.isBlank(atlasProperties.getProperty(TAGSYNC_ATLAS_CONSUMER_GROUP))) {
+ LOG.error("missing value for mandatory property '{}'", TAGSYNC_ATLAS_CONSUMER_GROUP);
+
+ ret = false;
+ }
+ }
+
+ return ret;
+ }
+
+ boolean validateRequiredAtlasKafkaProperties(Configuration atlasConfiguration) {
+ boolean ret = true;
+
+ if (atlasConfiguration == null) {
+ ret = false;
+ } else {
+ if (StringUtils.isBlank(atlasConfiguration.getString(TAGSYNC_ATLAS_KAFKA_ENDPOINTS))) {
+ LOG.error("missing value for mandatory property '{}'", TAGSYNC_ATLAS_KAFKA_ENDPOINTS);
+
+ ret = false;
+ }
+
+ if (StringUtils.isBlank(atlasConfiguration.getString(TAGSYNC_ATLAS_CONSUMER_GROUP))) {
+ LOG.error("missing value for mandatory property '{}'", TAGSYNC_ATLAS_CONSUMER_GROUP);
+
+ ret = false;
+ }
+ }
+
+ return ret;
+ }
+
+ private Configuration loadAtlasConfiguration() {
+ try {
+ ApplicationProperties.forceReload();
+
+ return ApplicationProperties.get();
+ } catch (AtlasException exception) {
+ LOG.error("Cannot load Atlas application properties file: {}", TAGSYNC_ATLAS_PROPERTIES_FILE_NAME, exception);
+
+ return null;
+ }
+ }
+
+ private static String getPrintableEntityNotification(EntityNotificationWrapper notification) {
+ StringBuilder sb = new StringBuilder();
+
+ sb.append("{ Notification-Type: ").append(notification.getOpType()).append(", ");
RangerAtlasEntityWithTags entityWithTags = new RangerAtlasEntityWithTags(notification);
- sb.append(entityWithTags.toString());
-
- sb.append("}");
- return sb.toString();
- }
-
- private class ConsumerRunnable implements Runnable {
-
- private final NotificationConsumer<EntityNotification> consumer;
-
- private final List<RangerAtlasEntityWithTags> atlasEntitiesWithTags = new ArrayList<>();
- private final List<AtlasKafkaMessage<EntityNotification>> messages = new ArrayList<>();
- private AtlasKafkaMessage<EntityNotification> lastUnhandledMessage = null;
-
- private long offsetOfLastMessageCommittedToKafka = -1L;
- private boolean isHandlingDeleteOps = false;
-
- private ConsumerRunnable(NotificationConsumer<EntityNotification> consumer) {
- this.consumer = consumer;
- }
-
- @Override
- public void run() {
- if (LOG.isDebugEnabled()) {
- LOG.debug("==> ConsumerRunnable.run()");
- }
-
- while (true) {
- if (TagSyncConfig.isTagSyncServiceActive()) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("==> ConsumerRunnable.run() is running as server is active");
- }
- try {
- List<AtlasKafkaMessage<EntityNotification>> newMessages = consumer.receive(MAX_WAIT_TIME_IN_MILLIS);
-
- if (newMessages.size() == 0) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("AtlasTagSource.ConsumerRunnable.run: no message from NotificationConsumer within " + MAX_WAIT_TIME_IN_MILLIS + " milliseconds");
- }
- if (CollectionUtils.isNotEmpty(atlasEntitiesWithTags)) {
- buildAndUploadServiceTags();
- }
- } else {
- for (AtlasKafkaMessage<EntityNotification> message : newMessages) {
- EntityNotification notification = message != null ? message.getMessage() : null;
-
- if (notification != null) {
- EntityNotificationWrapper notificationWrapper = null;
- try {
- notificationWrapper = new EntityNotificationWrapper(notification);
- } catch (Throwable e) {
- LOG.error("notification:[" + notification + "] has some issues..perhaps null entity??", e);
- }
- if (notificationWrapper != null) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Message-offset=" + message.getOffset() + ", Notification=" + getPrintableEntityNotification(notificationWrapper));
- }
-
- if (AtlasNotificationMapper.isNotificationHandled(notificationWrapper)) {
-
- if ((notificationWrapper.getIsEntityDeleteOp() && !isHandlingDeleteOps) || (!notificationWrapper.getIsEntityDeleteOp() && isHandlingDeleteOps)) {
- if (CollectionUtils.isNotEmpty(atlasEntitiesWithTags)) {
- buildAndUploadServiceTags();
- }
- isHandlingDeleteOps = !isHandlingDeleteOps;
- }
-
- atlasEntitiesWithTags.add(new RangerAtlasEntityWithTags(notificationWrapper));
- messages.add(message);
- } else {
- AtlasNotificationMapper.logUnhandledEntityNotification(notificationWrapper);
- lastUnhandledMessage = message;
- }
- }
- } else {
- LOG.error("Null entityNotification received from Kafka!! Ignoring..");
- }
- }
- if (CollectionUtils.isNotEmpty(atlasEntitiesWithTags) && atlasEntitiesWithTags.size() >= maxBatchSize) {
- buildAndUploadServiceTags();
- }
- }
- if (lastUnhandledMessage != null) {
- commitToKafka(lastUnhandledMessage);
- lastUnhandledMessage = null;
- }
-
- } catch (Exception exception) {
- LOG.error("Caught exception..: ", exception);
- // If transient error, retry after short interval
- try {
- Thread.sleep(100);
- } catch (InterruptedException interrupted) {
- LOG.error("Interrupted: ", interrupted);
- LOG.error("Returning from thread. May cause process to be up but not processing events!!");
- return;
- }
- }
- } else {
- try {
- long sleepInterval = TagSyncConfig.getTagSyncHAPassiveSleepInterval();
- LOG.debug("Sleeping for [{}] milliSeconds as this server is running in passive mode", sleepInterval);
- Thread.sleep(sleepInterval);
- } catch (InterruptedException interrupted) {
- LOG.error("Interrupted..: ", interrupted);
- // preserve interrupt status for caller of the thread
- Thread.currentThread().interrupt();
- return;
- }
- }
- }
- }
-
- private void buildAndUploadServiceTags() throws Exception {
- if (LOG.isDebugEnabled()) {
- LOG.debug("==> buildAndUploadServiceTags()");
- }
-
- if (CollectionUtils.isNotEmpty(atlasEntitiesWithTags) && CollectionUtils.isNotEmpty(messages)) {
-
- Map<String, ServiceTags> serviceTagsMap = AtlasNotificationMapper.processAtlasEntities(atlasEntitiesWithTags);
-
- if (MapUtils.isNotEmpty(serviceTagsMap)) {
- if (serviceTagsMap.size() != 1) {
- LOG.warn("Unexpected!! Notifications for more than one service received by AtlasTagSource.. Service-Names:[" + serviceTagsMap.keySet() + "]");
- }
- for (Map.Entry<String, ServiceTags> entry : serviceTagsMap.entrySet()) {
- if (isHandlingDeleteOps) {
- entry.getValue().setOp(ServiceTags.OP_DELETE);
- entry.getValue().setTagDefinitions(Collections.EMPTY_MAP);
- entry.getValue().setTags(Collections.EMPTY_MAP);
- } else {
- entry.getValue().setOp(ServiceTags.OP_ADD_OR_UPDATE);
- }
-
- if (LOG.isDebugEnabled()) {
- String serviceTagsString = JsonUtils.objectToJson(entry.getValue());
- LOG.debug("serviceTags=" + serviceTagsString);
- }
- updateSink(entry.getValue());
- }
- }
-
- AtlasKafkaMessage<EntityNotification> latestMessageDeliveredToRanger = messages.get(messages.size() - 1);
- commitToKafka(latestMessageDeliveredToRanger);
-
- atlasEntitiesWithTags.clear();
- messages.clear();
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Completed processing batch of messages of size:[" + messages.size() + "] received from NotificationConsumer");
- }
-
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("<== buildAndUploadServiceTags()");
- }
- }
-
- private void commitToKafka(AtlasKafkaMessage<EntityNotification> messageToCommit) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("==> commitToKafka(" + messageToCommit + ")");
- }
-
- long messageOffset = messageToCommit.getOffset();
- int partitionId = messageToCommit.getPartition();
-
- if (offsetOfLastMessageCommittedToKafka < messageOffset) {
- TopicPartition partition = new TopicPartition(messageToCommit.getTopic(), partitionId);
- try {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Committing message with offset:[" + messageOffset + "] to Kafka");
- }
- consumer.commit(partition, messageOffset);
- offsetOfLastMessageCommittedToKafka = messageOffset;
- } catch (Exception commitException) {
- LOG.warn("Ranger tagsync already processed message at offset " + messageOffset + ". Ignoring failure in committing message:[" + messageToCommit + "]", commitException);
- }
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("<== commitToKafka(" + messageToCommit + ")");
- }
- }
- }
+ sb.append(entityWithTags);
+
+ sb.append("}");
+ return sb.toString();
+ }
+
+ private class ConsumerRunnable implements Runnable {
+ private final NotificationConsumer<EntityNotification> consumer;
+
+ private final List<RangerAtlasEntityWithTags> atlasEntitiesWithTags = new ArrayList<>();
+ private final List<AtlasKafkaMessage<EntityNotification>> messages = new ArrayList<>();
+ private AtlasKafkaMessage<EntityNotification> lastUnhandledMessage;
+
+ private long offsetOfLastMessageCommittedToKafka = -1L;
+ private boolean isHandlingDeleteOps;
+
+ private ConsumerRunnable(NotificationConsumer<EntityNotification> consumer) {
+ this.consumer = consumer;
+ }
+
+ @Override
+ public void run() {
+ LOG.debug("==> ConsumerRunnable.run()");
+
+ while (true) {
+ if (TagSyncConfig.isTagSyncServiceActive()) {
+ LOG.debug("==> ConsumerRunnable.run() is running as server is active");
+ try {
+ List<AtlasKafkaMessage<EntityNotification>> newMessages = consumer.receive(MAX_WAIT_TIME_IN_MILLIS);
+
+ if (newMessages.isEmpty()) {
+ LOG.debug("AtlasTagSource.ConsumerRunnable.run: no message from NotificationConsumer within {} milliseconds", MAX_WAIT_TIME_IN_MILLIS);
+
+ if (CollectionUtils.isNotEmpty(atlasEntitiesWithTags)) {
+ buildAndUploadServiceTags();
+ }
+ } else {
+ for (AtlasKafkaMessage<EntityNotification> message : newMessages) {
+ EntityNotification notification = message != null ? message.getMessage() : null;
+
+ if (notification != null) {
+ EntityNotificationWrapper notificationWrapper = null;
+ try {
+ notificationWrapper = new EntityNotificationWrapper(notification);
+ } catch (Throwable e) {
+ LOG.error("notification:[{}] has some issues..perhaps null entity??", notification, e);
+ }
+ if (notificationWrapper != null) {
+ LOG.debug("Message-offset={}, Notification={}", message.getOffset(), getPrintableEntityNotification(notificationWrapper));
+
+ if (AtlasNotificationMapper.isNotificationHandled(notificationWrapper)) {
+ if ((notificationWrapper.getIsEntityDeleteOp() && !isHandlingDeleteOps) || (!notificationWrapper.getIsEntityDeleteOp() && isHandlingDeleteOps)) {
+ if (CollectionUtils.isNotEmpty(atlasEntitiesWithTags)) {
+ buildAndUploadServiceTags();
+ }
+ isHandlingDeleteOps = !isHandlingDeleteOps;
+ }
+
+ atlasEntitiesWithTags.add(new RangerAtlasEntityWithTags(notificationWrapper));
+ messages.add(message);
+ } else {
+ AtlasNotificationMapper.logUnhandledEntityNotification(notificationWrapper);
+ lastUnhandledMessage = message;
+ }
+ }
+ } else {
+ LOG.error("Null entityNotification received from Kafka!! Ignoring..");
+ }
+ }
+ if (CollectionUtils.isNotEmpty(atlasEntitiesWithTags) && atlasEntitiesWithTags.size() >= maxBatchSize) {
+ buildAndUploadServiceTags();
+ }
+ }
+ if (lastUnhandledMessage != null) {
+ commitToKafka(lastUnhandledMessage);
+ lastUnhandledMessage = null;
+ }
+ } catch (Exception exception) {
+ LOG.error("Caught exception..: ", exception);
+ // If transient error, retry after short interval
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException interrupted) {
+ LOG.error("Interrupted: ", interrupted);
+ LOG.error("Returning from thread. May cause process to be up but not processing events!!");
+ return;
+ }
+ }
+ } else {
+ try {
+ long sleepInterval = TagSyncConfig.getTagSyncHAPassiveSleepInterval();
+ LOG.debug("Sleeping for [{}] milliSeconds as this server is running in passive mode", sleepInterval);
+ Thread.sleep(sleepInterval);
+ } catch (InterruptedException interrupted) {
+ LOG.error("Interrupted: ", interrupted);
+ // preserve interrupt status for caller of the thread
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ }
+ }
+
+ private void buildAndUploadServiceTags() throws Exception {
+ LOG.debug("==> buildAndUploadServiceTags()");
+
+ if (CollectionUtils.isNotEmpty(atlasEntitiesWithTags) && CollectionUtils.isNotEmpty(messages)) {
+ Map<String, ServiceTags> serviceTagsMap = AtlasNotificationMapper.processAtlasEntities(atlasEntitiesWithTags);
+
+ if (MapUtils.isNotEmpty(serviceTagsMap)) {
+ if (serviceTagsMap.size() != 1) {
+ LOG.warn("Unexpected!! Notifications for more than one service received by AtlasTagSource.. Service-Names:[{}]", serviceTagsMap.keySet());
+ }
+ for (Map.Entry<String, ServiceTags> entry : serviceTagsMap.entrySet()) {
+ if (isHandlingDeleteOps) {
+ entry.getValue().setOp(ServiceTags.OP_DELETE);
+ entry.getValue().setTagDefinitions(Collections.EMPTY_MAP);
+ entry.getValue().setTags(Collections.EMPTY_MAP);
+ } else {
+ entry.getValue().setOp(ServiceTags.OP_ADD_OR_UPDATE);
+ }
+
+ LOG.debug("serviceTags= {}", JsonUtils.objectToJson(entry.getValue()));
+
+ updateSink(entry.getValue());
+ }
+ }
+
+ AtlasKafkaMessage<EntityNotification> latestMessageDeliveredToRanger = messages.get(messages.size() - 1);
+ commitToKafka(latestMessageDeliveredToRanger);
+
+ atlasEntitiesWithTags.clear();
+ messages.clear();
+
+ LOG.debug("Completed processing batch of messages of size:[{}] received from NotificationConsumer", messages.size());
+ }
+
+ LOG.debug("<== buildAndUploadServiceTags()");
+ }
+
+ private void commitToKafka(AtlasKafkaMessage<EntityNotification> messageToCommit) {
+ LOG.debug("==> commitToKafka({})", messageToCommit);
+
+ long messageOffset = messageToCommit.getOffset();
+ int partitionId = messageToCommit.getPartition();
+
+ if (offsetOfLastMessageCommittedToKafka < messageOffset) {
+ TopicPartition partition = new TopicPartition(messageToCommit.getTopic(), partitionId);
+ try {
+ LOG.debug("Committing message with offset:[{}] to Kafka", messageOffset);
+ consumer.commit(partition, messageOffset);
+ offsetOfLastMessageCommittedToKafka = messageOffset;
+ } catch (Exception commitException) {
+ LOG.warn("Ranger tagsync already processed message at offset {}. Ignoring failure in committing message:[{}]", messageOffset, messageToCommit, commitException);
+ }
+ }
+
+ LOG.debug("<== commitToKafka({})", messageToCommit);
+ }
+ }
}
-
diff --git a/tagsync/src/test/java/org/apache/ranger/tagsync/source/atlas/AtlasTagSourceConfigTest.java b/tagsync/src/test/java/org/apache/ranger/tagsync/source/atlas/AtlasTagSourceConfigTest.java
new file mode 100644
index 000000000..37285751f
--- /dev/null
+++ b/tagsync/src/test/java/org/apache/ranger/tagsync/source/atlas/AtlasTagSourceConfigTest.java
@@ -0,0 +1,57 @@
+/*
+ * 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.atlas;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Properties;
+
+public class AtlasTagSourceConfigTest {
+ private final AtlasTagSource tagSource = new AtlasTagSource();
+
+ @Test
+ public void validateRequiredAtlasKafkaProperties_acceptsBootstrapAndGroupWithoutZookeeper() {
+ Properties atlasProperties = new Properties();
+
+ atlasProperties.setProperty(AtlasTagSource.TAGSYNC_ATLAS_KAFKA_ENDPOINTS, "kafka1:9092,kafka2:9092");
+ atlasProperties.setProperty(AtlasTagSource.TAGSYNC_ATLAS_CONSUMER_GROUP, "ranger_entities_consumer");
+
+ Assertions.assertTrue(tagSource.validateRequiredAtlasKafkaProperties(atlasProperties));
+ }
+
+ @Test
+ public void validateRequiredAtlasKafkaProperties_rejectsMissingBootstrapServers() {
+ Properties atlasProperties = new Properties();
+
+ atlasProperties.setProperty(AtlasTagSource.TAGSYNC_ATLAS_CONSUMER_GROUP, "ranger_entities_consumer");
+
+ Assertions.assertFalse(tagSource.validateRequiredAtlasKafkaProperties(atlasProperties));
+ }
+
+ @Test
+ public void validateRequiredAtlasKafkaProperties_rejectsMissingConsumerGroup() {
+ Properties atlasProperties = new Properties();
+
+ atlasProperties.setProperty(AtlasTagSource.TAGSYNC_ATLAS_KAFKA_ENDPOINTS, "kafka1:9092");
+
+ Assertions.assertFalse(tagSource.validateRequiredAtlasKafkaProperties(atlasProperties));
+ }
+}