This is an automated email from the ASF dual-hosted git repository. ramackri pushed a commit to branch RANGER-5658-backport-ranger-2.9 in repository https://gitbox.apache.org/repos/asf/ranger.git
commit 8e981ce471b257a2c5a864cfb08c6e8a698bcade Author: ramk <[email protected]> AuthorDate: Sat Jul 4 21:28:18 2026 +0530 RANGER-5658: Tag Sync: remove atlas.kafka.zookeeper.connect and fix Atlas Kafka consumer for Kafka 3.9 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 --- 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)); + } +}
