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 <[email protected]>
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 <[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 | 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/[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 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/[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 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));
+ }
+}
