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 deaf13df3 RANGER-5658: Tag Sync: remove atlas.kafka.zookeeper.connect
and fix Atlas Kafka consumer for Kafka 3.9 (#1037)
deaf13df3 is described below
commit deaf13df3c62af8edde53e185890682fb711bec5
Author: Ramachandran Krishnan <[email protected]>
AuthorDate: Fri Jul 3 22:38:01 2026 +0530
RANGER-5658: Tag Sync: remove atlas.kafka.zookeeper.connect and fix Atlas
Kafka consumer for Kafka 3.9 (#1037)
* RANGER-5658: Remove obsolete atlas.kafka.zookeeper.connect requirement
from Tag Sync.
Tag Sync consumes Atlas notifications via kafka-clients (bootstrap.servers
only).
The legacy zookeeper.connect property was never used by Atlas
KafkaNotification
but was still required at startup and in installer templates.
Co-authored-by: Cursor <[email protected]>
* RANGER-5658: Remove dev-support PR template README from the patch.
Co-authored-by: Cursor <[email protected]>
* TagSync install: use rangerTagsync_password for Ranger jceks credential.
setup.py wrote tagadmin.user.password as the literal username
'rangertagsync'
instead of rangerTagsync_password from install.properties, causing 401 on
TagAdminRESTSink until updatetagadminpassword.py ran at end of setup.
Complements RANGER-5658 / PR #1037 (Atlas Kafka ZK cleanup); required for
docker Atlas REST TagSync with Kerberos Ranger Admin.
Co-authored-by: Cursor <[email protected]>
* RANGER-5658: Enable Tag Sync Atlas Kafka consumer for PLAINTEXT and KRaft.
Emit Kerberos JAAS in setup.py only when atlas.kafka.security.protocol is
SASL; load atlas-application.properties via ApplicationProperties and
-Datlas.conf; update docker sample to atlas-kafka PLAINTEXT bootstrap.
Co-authored-by: Cursor <[email protected]>
* RANGER-5658: Add Tag Sync Atlas Kafka operator documentation.
Document PLAINTEXT and Kerberos Kafka modes, startup commands,
-Datlas.conf requirement, and verification steps.
Co-authored-by: Cursor <[email protected]>
* RANGER-5658: Align TagSync docs and docker sample with ranger-docker
Kafka.
Use ranger-kafka.rangernw:9092 with SASL_PLAINTEXT in docker install sample;
restore localhost Atlas REST endpoint; trim README change-log section.
Co-authored-by: Cursor <[email protected]>
* RANGER-5658: Default ranger-docker TagSync to Atlas REST source.
Keep Atlas Kafka settings in install sample for optional enablement;
document REST-as-default in README.
Co-authored-by: Cursor <[email protected]>
* RANGER-5658: Fix AtlasTagSource import order for checkstyle.
Order commons.collections imports before commons.configuration per
dev-support/checkstyle.xml ImportOrder rules.
Co-authored-by: Cursor <[email protected]>
* Update AtlasTagSource.java
* RANGER-5658: Restore ranger-docker TagSync file-source default per review.
dev-support/ranger-docker has no Atlas server; default to file source
(TAG_SOURCE_FILE_ENABLED=true) instead of Atlas REST. Document coexist
Kafka enablement in README.
* RANGER-5658: Fix AtlasTagSource trailing whitespace for checkstyle.
Remove trailing spaces on blank lines reported by CI checkstyle
(RegexpMultiline) in AtlasTagSource.java.
---------
Co-authored-by: ramk <[email protected]>
Co-authored-by: Cursor <[email protected]>
Co-authored-by: Madhan Neethiraj <[email protected]>
---
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 | 131 +++++++------
.../source/atlas/AtlasTagSourceConfigTest.java | 57 ++++++
8 files changed, 367 insertions(+), 70 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/[email protected]
+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 88d040495..c34d74ee7 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.rangernw: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/[email protected]
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 eeeccaf4a..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,6 +19,8 @@
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;
@@ -26,6 +28,7 @@
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;
@@ -36,8 +39,6 @@
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;
@@ -48,12 +49,11 @@ 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;
+ 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;
@@ -61,56 +61,22 @@ public class AtlasTagSource extends AbstractTagSource {
public boolean initialize(Properties properties) {
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");
- }
- }
+ Configuration atlasConfiguration = loadAtlasConfiguration();
- if (ret) {
- if
(StringUtils.isBlank(atlasProperties.getProperty(TAGSYNC_ATLAS_KAFKA_ENDPOINTS)))
{
- ret = false;
- LOG.error("Value of property '{}' is not specified!",
TAGSYNC_ATLAS_KAFKA_ENDPOINTS);
- }
- if
(StringUtils.isBlank(atlasProperties.getProperty(TAGSYNC_ATLAS_ZOOKEEPER_ENDPOINT)))
{
- ret = false;
- LOG.error("Value of property '{}' is not specified!",
TAGSYNC_ATLAS_ZOOKEEPER_ENDPOINT);
- }
- if
(StringUtils.isBlank(atlasProperties.getProperty(TAGSYNC_ATLAS_CONSUMER_GROUP)))
{
- ret = false;
- LOG.error("Value of property '{}' is not specified!",
TAGSYNC_ATLAS_CONSUMER_GROUP);
- }
- }
+ ret = validateRequiredAtlasKafkaProperties(atlasConfiguration);
- if (ret) {
- NotificationInterface notification =
NotificationProvider.get();
- List<NotificationConsumer<EntityNotification>> iterators =
notification.createConsumers(NotificationInterface.NotificationType.ENTITIES,
1);
+ 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);
+ consumerTask = new ConsumerRunnable(iterators.get(0));
+ }
}
- maxBatchSize = TagSyncConfig.getSinkMaxBatchSize(properties);
-
LOG.debug("<== AtlasTagSource.initialize(), result={}", ret);
return ret;
@@ -120,15 +86,22 @@ public boolean initialize(Properties properties) {
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()");
- return myThread != null;
+
+ LOG.debug("<== AtlasTagSource.start(): ret={}", ret);
+
+ return ret;
}
@Override
@@ -138,6 +111,62 @@ public void stop() {
}
}
+ 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();
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));
+ }
+}
