Repository: sentry Updated Branches: refs/heads/SENTRY-1874 [created] 10e65dadf
SENTRY-1874: Do not require quiet HMS when taking initial HMS snapshot Project: http://git-wip-us.apache.org/repos/asf/sentry/repo Commit: http://git-wip-us.apache.org/repos/asf/sentry/commit/10e65dad Tree: http://git-wip-us.apache.org/repos/asf/sentry/tree/10e65dad Diff: http://git-wip-us.apache.org/repos/asf/sentry/diff/10e65dad Branch: refs/heads/SENTRY-1874 Commit: 10e65dadfe2a638e08be538d3ff196bff2aeda3d Parents: 9681099 Author: Alexander Kolbasov <ak...@cloudera.com> Authored: Fri Aug 11 14:24:37 2017 -0700 Committer: Alexander Kolbasov <ak...@cloudera.com> Committed: Fri Aug 11 14:24:37 2017 -0700 ---------------------------------------------------------------------- .../apache/sentry/hdfs/DBUpdateForwarder.java | 2 +- .../service/thrift/FullUpdateInitializer.java | 2 +- .../service/thrift/FullUpdateModifier.java | 605 +++++++++++++++++++ .../sentry/service/thrift/HMSFollower.java | 19 +- .../sentry/service/thrift/SentryHMSClient.java | 72 ++- .../service/thrift/TestFullUpdateModifier.java | 453 ++++++++++++++ 6 files changed, 1126 insertions(+), 27 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/sentry/blob/10e65dad/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/DBUpdateForwarder.java ---------------------------------------------------------------------- diff --git a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/DBUpdateForwarder.java b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/DBUpdateForwarder.java index 1ab4d6f..e4664af 100644 --- a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/DBUpdateForwarder.java +++ b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/DBUpdateForwarder.java @@ -105,7 +105,7 @@ class DBUpdateForwarder<K extends Updateable.Update> { if (seqNum > SEQUENCE_NUMBER_UPDATE_UNINITIALIZED && deltaRetriever.isDeltaAvailable(seqNum)) { List<K> deltas = deltaRetriever.retrieveDelta(seqNum); if (!deltas.isEmpty()) { - LOGGER.info("Newer delta updates are found up to sequence number: ", curSeqNum); + LOGGER.info("Newer delta updates are found up to sequence number: {}", curSeqNum); return deltas; } } http://git-wip-us.apache.org/repos/asf/sentry/blob/10e65dad/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/FullUpdateInitializer.java ---------------------------------------------------------------------- diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/FullUpdateInitializer.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/FullUpdateInitializer.java index 5af1e4f..760a2b5 100644 --- a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/FullUpdateInitializer.java +++ b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/FullUpdateInitializer.java @@ -128,7 +128,7 @@ public final class FullUpdateInitializer implements AutoCloseable { * @param uri - resource URI (usually with scheme) * @return path if uri is valid or null */ - private static String pathFromURI(String uri) { + static String pathFromURI(String uri) { try { return PathsUpdate.parsePath(uri); } catch (SentryMalformedPathException e) { http://git-wip-us.apache.org/repos/asf/sentry/blob/10e65dad/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/FullUpdateModifier.java ---------------------------------------------------------------------- diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/FullUpdateModifier.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/FullUpdateModifier.java new file mode 100644 index 0000000..2cd18ea --- /dev/null +++ b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/FullUpdateModifier.java @@ -0,0 +1,605 @@ +/* + 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.sentry.service.thrift; + +import com.google.common.annotations.VisibleForTesting; +import org.apache.hadoop.hive.metastore.api.NotificationEvent; +import org.apache.hive.hcatalog.messaging.HCatEventMessage; +import org.apache.hive.hcatalog.messaging.MessageDeserializer; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONAddPartitionMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONAlterPartitionMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONAlterTableMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONCreateDatabaseMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONCreateTableMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONDropDatabaseMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONDropPartitionMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONDropTableMessage; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Apply newer events to the full update. + * + * <p>The process of obtaining ful snapshot from HMS is not atomic. + * While we read information from HMS it may change - some new objects can be created, + * or some can be removed or modified. This class is used to reconsile changes to + * the full snapshot. + */ +final class FullUpdateModifier { + private static final Logger LOGGER = LoggerFactory.getLogger(FullUpdateModifier.class); + + // Prevent creation of class instances + private FullUpdateModifier() { + } + + /** + * Take a full snapshot and apply an MS event to it. + * + * <p>We pass serializer as a parameter to simplify testing. + * + * @param image Full snapshot + * @param event HMS notificatin event + * @param deserializer Message deserializer - + * should produce Sentry JSON serializer type messages. + */ + // NOTE: we pass deserializer here instead of using built-in one to simplify testing. + // Tests use mock serializers and thus we do not have to construct proper events. + static void applyEvent(Map<String, Set<String>> image, NotificationEvent event, + MessageDeserializer deserializer) { + HCatEventMessage.EventType eventType = + HCatEventMessage.EventType.valueOf(event.getEventType()); + + switch (eventType) { + case CREATE_DATABASE: + createDatabase(image, event, deserializer); + break; + case DROP_DATABASE: + dropDatabase(image, event, deserializer); + break; + case CREATE_TABLE: + createTable(image, event, deserializer); + break; + case DROP_TABLE: + dropTable(image, event, deserializer); + break; + case ALTER_TABLE: + alterTable(image, event, deserializer); + break; + case ADD_PARTITION: + addPartition(image, event, deserializer); + break; + case DROP_PARTITION: + dropPartition(image, event, deserializer); + break; + case ALTER_PARTITION: + alterPartition(image, event, deserializer); + break; + default: + LOGGER.error("Notification with ID:{} has invalid event type: {}", event.getEventId(), + event.getEventType()); + break; + } + } + + /** + * Add mapping from the new database name to location {dbname: {location}}. + */ + private static void createDatabase(Map<String, Set<String>> image, NotificationEvent event, + MessageDeserializer deserializer) { + SentryJSONCreateDatabaseMessage message = + (SentryJSONCreateDatabaseMessage) deserializer + .getCreateDatabaseMessage(event.getMessage()); + + String dbName = message.getDB(); + if ((dbName == null) || dbName.isEmpty()) { + LOGGER.error("Create database event is missing database name"); + return; + } + dbName = dbName.toLowerCase(); + + String location = message.getLocation(); + if ((location == null) || location.isEmpty()) { + LOGGER.error("Create database event is missing database location"); + return; + } + + String path = FullUpdateInitializer.pathFromURI(location); + if (path == null) { + return; + } + + // Add new database if it doesn't exist yet + if (!image.containsKey(dbName)) { + LOGGER.debug("create database {} with location {}", dbName, location); + image.put(dbName.intern(), Collections.singleton(path)); + } else { + // Sanity check the information and print warnings if database exists but + // with a different location + Set<String> oldLocations = image.get(dbName); + LOGGER.debug("database {} already exists, ignored", dbName); + if (!oldLocations.contains(location)) { + LOGGER.warn("database {} exists but location is different from {}", dbName, location); + } + } + } + + /** + * Remove a mapping from database name and remove all mappings which look like dbName.tableName + * where dbName matches database name. + */ + private static void dropDatabase(Map<String, Set<String>> image, NotificationEvent event, + MessageDeserializer deserializer) { + SentryJSONDropDatabaseMessage message = + (SentryJSONDropDatabaseMessage) deserializer.getDropDatabaseMessage(event.getMessage()); + + String dbName = message.getDB(); + if ((dbName == null) || dbName.isEmpty()) { + LOGGER.error("Drop database event is missing database name"); + return; + } + dbName = dbName.toLowerCase(); + String location = message.getLocation(); + if ((location == null) || location.isEmpty()) { + LOGGER.error("Drop database event is missing database location"); + return; + } + + String path = FullUpdateInitializer.pathFromURI(location); + if (path == null) { + return; + } + + // If the database is alreday deleted, we have nothing to do + Set<String> locations = image.get(dbName); + if (locations == null) { + LOGGER.debug("database {} is already deleted", dbName); + return; + } + + if (!locations.contains(path)) { + LOGGER.warn("Database {} location does not match {}", dbName, path); + return; + } + + LOGGER.debug("drop database {} with location {}", dbName, location); + + // Drop information about the database + image.remove(dbName); + + String dbPrefix = dbName + "."; + + // Remove all objects for this database + for (Iterator<Map.Entry<String, Set<String>>> it = image.entrySet().iterator(); + it.hasNext(); ) { + Map.Entry<String, Set<String>> entry = it.next(); + String key = entry.getKey(); + if (key.startsWith(dbPrefix)) { + LOGGER.debug("Removing {}", key); + it.remove(); + } + } + } + + /** + * Add mapping for dbName.tableName. + */ + private static void createTable(Map<String, Set<String>> image, NotificationEvent event, + MessageDeserializer deserializer) { + SentryJSONCreateTableMessage message = (SentryJSONCreateTableMessage) deserializer + .getCreateTableMessage(event.getMessage()); + + String dbName = message.getDB(); + if ((dbName == null) || dbName.isEmpty()) { + LOGGER.error("Create table event is missing database name"); + return; + } + String tableName = message.getTable(); + if ((tableName == null) || tableName.isEmpty()) { + LOGGER.error("Create table event is missing table name"); + return; + } + + String location = message.getLocation(); + if ((location == null) || location.isEmpty()) { + LOGGER.error("Create table event is missing table location"); + return; + } + + String path = FullUpdateInitializer.pathFromURI(location); + if (path == null) { + return; + } + + String authName = dbName.toLowerCase() + "." + tableName.toLowerCase(); + // Add new table if it doesn't exist yet + if (!image.containsKey(authName)) { + LOGGER.debug("create table {} with location {}", authName, location); + Set<String> locations = new HashSet<>(1); + locations.add(path); + image.put(authName.intern(), locations); + } else { + // Sanity check the information and print warnings if table exists but + // with a different location + Set<String> oldLocations = image.get(authName); + LOGGER.debug("Table {} already exists, ignored", authName); + if (!oldLocations.contains(location)) { + LOGGER.warn("Table {} exists but location is different from {}", authName, location); + } + } + } + + /** + * Drop mapping from dbName.tableName + */ + private static void dropTable(Map<String, Set<String>> image, NotificationEvent event, + MessageDeserializer deserializer) { + SentryJSONDropTableMessage message = (SentryJSONDropTableMessage) deserializer + .getDropTableMessage(event.getMessage()); + + String dbName = message.getDB(); + if ((dbName == null) || dbName.isEmpty()) { + LOGGER.error("Drop table event is missing database name"); + return; + } + String tableName = message.getTable(); + if ((tableName == null) || tableName.isEmpty()) { + LOGGER.error("Drop table event is missing table name"); + return; + } + + String location = message.getLocation(); + if ((location == null) || location.isEmpty()) { + LOGGER.error("Drop table event is missing table location"); + return; + } + + String path = FullUpdateInitializer.pathFromURI(location); + if (path == null) { + return; + } + + String authName = dbName.toLowerCase() + "." + tableName.toLowerCase(); + Set<String> locations = image.get(authName); + if (locations != null && locations.contains(path)) { + LOGGER.debug("Removing {}", authName); + image.remove(authName); + } else { + LOGGER.warn("can't find matching table {} with location {}", authName, location); + } + } + + /** + * ALTER TABLE is a complicated function that can alter multiple things. + * + * <p>We take care iof the following cases: + * <ul> + * <li>Change database name. This is the most complicated one. + * We need to change the actual database name and change all mappings + * that look like "dbName.tableName" to the new dbName</li> + * <li>Change table name</li> + * <li>Change location</li> + * </ul> + * + */ + private static void alterTable(Map<String, Set<String>> image, NotificationEvent event, + MessageDeserializer deserializer) { + SentryJSONAlterTableMessage message = + (SentryJSONAlterTableMessage) deserializer.getAlterTableMessage(event.getMessage()); + String prevDbName = message.getDB(); + if ((prevDbName == null) || prevDbName.isEmpty()) { + LOGGER.error("Alter table event is missing old database name"); + return; + } + prevDbName = prevDbName.toLowerCase(); + String prevTableName = message.getTable(); + if ((prevTableName == null) || prevTableName.isEmpty()) { + LOGGER.error("Alter table event is missing old table name"); + return; + } + prevTableName = prevTableName.toLowerCase(); + + String newDbName = event.getDbName(); + if ((newDbName == null) || newDbName.isEmpty()) { + LOGGER.error("Alter table event is missing new database name"); + return; + } + newDbName = newDbName.toLowerCase(); + + String newTableName = event.getTableName(); + if ((newTableName == null) || newTableName.isEmpty()) { + LOGGER.error("Alter table event is missing new table name"); + return; + } + newTableName = newTableName.toLowerCase(); + + String prevLocation = message.getOldLocation(); + if ((prevLocation == null) || prevLocation.isEmpty()) { + LOGGER.error("Alter table event is missing old location"); + return; + } + String prevPath = FullUpdateInitializer.pathFromURI(prevLocation); + if (prevPath == null) { + return; + } + + String newLocation = message.getNewLocation(); + if ((newLocation == null) || newLocation.isEmpty()) { + LOGGER.error("Alter table event is missing new location"); + return; + } + String newPath = FullUpdateInitializer.pathFromURI(newLocation); + if (newPath == null) { + return; + } + + String prevAuthName = prevDbName + "." + prevTableName; + String newAuthName = newDbName + "." + newTableName; + + if (!prevDbName.equals(newDbName)) { + // Database name change + LOGGER.debug("Changing database name: {} -> {}", prevDbName, newDbName); + Set<String> locations = image.get(prevDbName); + if (locations != null) { + // Rename database if it is not renamed yet + if (!image.containsKey(newDbName)) { + image.put(newDbName, locations); + image.remove(prevDbName); + // Walk through all tables and rename DB part of the AUTH name + // AUTH name is "dbName.TableName" so we need to replace dbName with the new name + String prevDbPrefix = prevDbName + "."; + String newDbPrefix = newDbName + "."; + renamePrefixKeys(image, prevDbPrefix, newDbPrefix); + } else { + LOGGER.warn("database {} rename: found existing database {}", prevDbName, newDbName); + } + } else { + LOGGER.debug("database {} not found", prevDbName); + } + } + + if (!prevAuthName.equals(newAuthName)) { + // Either the database name or table name changed, rename objects + Set<String> locations = image.get(prevAuthName); + if (locations != null) { + // Rename if it is not renamed yet + if (!image.containsKey(newAuthName)) { + LOGGER.debug("rename {} -> {}", prevAuthName, newAuthName); + image.put(newAuthName, locations); + image.remove(prevAuthName); + } else { + LOGGER.warn("auth {} rename: found existing object {}", prevAuthName, newAuthName); + } + } else { + LOGGER.debug("auth {} not found", prevAuthName); + } + } + + if (!prevPath.equals(newPath)) { + LOGGER.debug("Location change: {} -> {}", prevPath, newPath); + // Location change + Set<String> locations = image.get(newAuthName); + if (locations != null && locations.contains(prevPath) && !locations.contains(newPath)) { + locations.remove(prevPath); + locations.add(newPath); + } else { + LOGGER.warn("can not process location change for {}", newAuthName); + LOGGER.warn("old locatio = {}, new location = {}", prevPath, newPath); + } + } + } + + /** + * Add partition just adds a new location to the existing table. + */ + private static void addPartition(Map<String, Set<String>> image, NotificationEvent event, + MessageDeserializer deserializer) { + SentryJSONAddPartitionMessage message = + (SentryJSONAddPartitionMessage) deserializer.getAddPartitionMessage(event.getMessage()); + + String dbName = message.getDB(); + if ((dbName == null) || dbName.isEmpty()) { + LOGGER.error("Add partition event is missing database name"); + return; + } + String tableName = message.getTable(); + if ((tableName == null) || tableName.isEmpty()) { + LOGGER.error("Add partition event for {} is missing table name", dbName); + return; + } + + String authName = dbName.toLowerCase() + "." + tableName.toLowerCase(); + + List<String> locations = message.getLocations(); + if (locations == null || locations.isEmpty()) { + LOGGER.error("Add partition event for {} is missing partition locations", authName); + return; + } + + Set<String> oldLocations = image.get(authName); + if (oldLocations == null) { + LOGGER.warn("Add partition for {}: missing table locations",authName); + return; + } + + // Add each partition + for (String location: locations) { + String path = FullUpdateInitializer.pathFromURI(location); + if (path != null) { + LOGGER.debug("Adding partition {}:{}", authName, path); + oldLocations.add(path); + } + } + } + + /** + * Drop partition removes location from the existing table. + */ + private static void dropPartition(Map<String, Set<String>> image, NotificationEvent event, + MessageDeserializer deserializer) { + SentryJSONDropPartitionMessage message = + (SentryJSONDropPartitionMessage) deserializer + .getDropPartitionMessage(event.getMessage()); + String dbName = message.getDB(); + if ((dbName == null) || dbName.isEmpty()) { + LOGGER.error("Drop partition event is missing database name"); + return; + } + String tableName = message.getTable(); + if ((tableName == null) || tableName.isEmpty()) { + LOGGER.error("Drop partition event for {} is missing table name", dbName); + return; + } + + String authName = dbName.toLowerCase() + "." + tableName.toLowerCase(); + + List<String> locations = message.getLocations(); + if (locations == null || locations.isEmpty()) { + LOGGER.error("Drop partition event for {} is missing partition locations", authName); + return; + } + + Set<String> oldLocations = image.get(authName); + if (oldLocations == null) { + LOGGER.warn("Add partition for {}: missing table locations",authName); + return; + } + + // Drop each partition + for (String location: locations) { + String path = FullUpdateInitializer.pathFromURI(location); + if (path != null) { + oldLocations.remove(path); + } + } + } + + private static void alterPartition(Map<String, Set<String>> image, NotificationEvent event, + MessageDeserializer deserializer) { + SentryJSONAlterPartitionMessage message = + (SentryJSONAlterPartitionMessage) deserializer + .getAlterPartitionMessage(event.getMessage()); + + String dbName = message.getDB(); + if ((dbName == null) || dbName.isEmpty()) { + LOGGER.error("Alter partition event is missing database name"); + return; + } + String tableName = message.getTable(); + if ((tableName == null) || tableName.isEmpty()) { + LOGGER.error("Alter partition event for {} is missing table name", dbName); + return; + } + + String authName = dbName.toLowerCase() + "." + tableName.toLowerCase(); + + String prevLocation = message.getOldLocation(); + if (prevLocation == null || prevLocation.isEmpty()) { + LOGGER.error("Alter partition event for {} is missing old location", authName); + } + + String prevPath = FullUpdateInitializer.pathFromURI(prevLocation); + if (prevPath == null) { + return; + } + + String newLocation = message.getNewLocation(); + if (newLocation == null || newLocation.isEmpty()) { + LOGGER.error("Alter partition event for {} is missing new location", authName); + } + + String newPath = FullUpdateInitializer.pathFromURI(newLocation); + if (newPath == null) { + return; + } + + if (prevPath.equals(newPath)) { + LOGGER.warn("Alter partition event for {} has the same old and new path {}", + authName, prevPath); + return; + } + + Set<String> locations = image.get(authName); + if (locations == null) { + LOGGER.warn("Missing partition locations for {}", authName); + return; + } + + // Rename partition + if (locations.remove(prevPath)) { + LOGGER.debug("Renaming {} to {}", prevPath, newPath); + locations.add(newPath); + } + } + + /** + * Walk through the map and rename all instances of oldKey to newKey. + */ + @VisibleForTesting + protected static void renamePrefixKeys(Map<String, Set<String>> image, + String oldKey, String newKey) { + // The trick is that we can't just iterate through the map, remove old values and + // insert new values. While we can remove old values with iterators, + // we can't insert new ones while we walk. So we collect the keys to be added in + // a new map and merge them in the end. + Map<String, Set<String>> replacement = new HashMap<>(); + + for (Iterator<Map.Entry<String, Set<String>>> it = image.entrySet().iterator(); + it.hasNext(); ) { + Map.Entry<String, Set<String>> entry = it.next(); + String key = entry.getKey(); + if (key.startsWith(oldKey)) { + String updatedKey = key.replaceAll("^" + oldKey + "(.*)", newKey + "$1"); + if (!image.containsKey(updatedKey)) { + LOGGER.debug("Rename {} to {}", key, updatedKey); + replacement.put(updatedKey, entry.getValue()); + it.remove(); + } else { + LOGGER.warn("skipping key {} - already present", updatedKey); + } + } + } + + mergeMaps(image, replacement); + } + + /** + * Merge replacement values into the original map but only if they are not + * already there. + * + * @param m1 source map + * @param m2 map with replacement values + */ + private static void mergeMaps(Map<String, Set<String>> m1, Map<String, Set<String>> m2) { + // Merge replacement values into the original map but only if they are not + // already there + for (Map.Entry<String, Set<String>> entry : m2.entrySet()) { + if (!m1.containsKey(entry.getKey())) { + m1.put(entry.getKey(), entry.getValue()); + } + } + } +} http://git-wip-us.apache.org/repos/asf/sentry/blob/10e65dad/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/HMSFollower.java ---------------------------------------------------------------------- diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/HMSFollower.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/HMSFollower.java index b0a202e..53a3fa4 100644 --- a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/HMSFollower.java +++ b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/HMSFollower.java @@ -6,9 +6,9 @@ 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 - <p> + http://www.apache.org/licenses/LICENSE-2.0 - <p> + 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. @@ -64,7 +64,6 @@ public class HMSFollower implements Runnable, AutoCloseable { this(conf, store, leaderMonitor, hiveConnectionFactory, null); } - @VisibleForTesting /** * Constructor should be used only for testing purposes. * @@ -73,6 +72,7 @@ public class HMSFollower implements Runnable, AutoCloseable { * @param leaderMonitor * @param authServerName Server that sentry is Authorizing */ + @VisibleForTesting public HMSFollower(Configuration conf, SentryStore store, LeaderStatusMonitor leaderMonitor, HiveSimpleConnectionFactory hiveConnectionFactory, String authServerName) { LOGGER.info("HMSFollower is being initialized"); @@ -244,12 +244,13 @@ public class HMSFollower implements Runnable, AutoCloseable { long firstNotificationId = eventList.get(0).getEventId(); long lastNotificationId = eventList.get(eventList.size() - 1).getEventId(); - /* If the next expected notification is not available, then an out-of-sync might - * have happened due to the following issue: - * - * - HDFS sync was disabled or Sentry was shutdown for a time period longer than - * the HMS notification clean-up thread causing old notifications to be deleted. - */ + // + // If the next expected notification is not available, then an out-of-sync might + // have happened due to the following issue: + // + // - HDFS sync was disabled or Sentry was shutdown for a time period longer than + // the HMS notification clean-up thread causing old notifications to be deleted. + // if ((latestProcessedId + 1) != firstNotificationId) { LOGGER.info("Current HMS notifications are out-of-sync with latest Sentry processed" + "notifications. Need to request a full HMS snapshot."); http://git-wip-us.apache.org/repos/asf/sentry/blob/10e65dad/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/SentryHMSClient.java ---------------------------------------------------------------------- diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/SentryHMSClient.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/SentryHMSClient.java index 4f76a94..1d91fc2 100644 --- a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/SentryHMSClient.java +++ b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/SentryHMSClient.java @@ -19,6 +19,7 @@ package org.apache.sentry.service.thrift; import static com.codahale.metrics.MetricRegistry.name; +import static java.util.Collections.emptyMap; import com.codahale.metrics.Counter; import com.codahale.metrics.Timer; @@ -39,6 +40,7 @@ import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId; import org.apache.hadoop.hive.metastore.api.MetaException; import org.apache.hadoop.hive.metastore.api.NotificationEvent; import org.apache.hadoop.hive.metastore.api.NotificationEventResponse; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONMessageDeserializer; import org.apache.sentry.provider.db.service.persistent.PathsImage; import org.apache.sentry.provider.db.service.persistent.SentryStore; import org.apache.sentry.provider.db.service.thrift.SentryMetrics; @@ -57,6 +59,8 @@ import org.slf4j.LoggerFactory; class SentryHMSClient implements AutoCloseable { private static final Logger LOGGER = LoggerFactory.getLogger(SentryHMSClient.class); + private static final String NOT_CONNECTED_MSG = "Client is not connected to HMS"; + private final Configuration conf; private HiveMetaStoreClient client = null; private HiveConnectionFactory hiveConnectionFactory; @@ -142,7 +146,7 @@ class SentryHMSClient implements AutoCloseable { PathsImage getFullSnapshot() { try { if (client == null) { - LOGGER.error("Client is not connected to HMS"); + LOGGER.error(NOT_CONNECTED_MSG); return new PathsImage(Collections.<String, Set<String>>emptyMap(), SentryStore.EMPTY_NOTIFICATION_ID, SentryStore.EMPTY_PATHS_SNAPSHOT_ID); } @@ -157,22 +161,58 @@ class SentryHMSClient implements AutoCloseable { CurrentNotificationEventId eventIdAfter = client.getCurrentNotificationEventId(); LOGGER.info("NotificationID, Before Snapshot: {}, After Snapshot {}", eventIdBefore.getEventId(), eventIdAfter.getEventId()); - // To ensure point-in-time snapshot consistency, need to make sure - // there were no HMS updates while retrieving the snapshot. If there are updates, snapshot - // is discarded. New attempt will be made after 500 milliseconds when - // HMSFollower runs again. - if (!eventIdBefore.equals(eventIdAfter)) { - LOGGER.error("Snapshot discarded, updates to HMS data while shapshot is being taken." - + "ID Before: {}. ID After: {}", eventIdBefore.getEventId(), eventIdAfter.getEventId()); - return new PathsImage(Collections.<String, Set<String>>emptyMap(), - SentryStore.EMPTY_NOTIFICATION_ID, SentryStore.EMPTY_PATHS_SNAPSHOT_ID); + + if (eventIdAfter.equals(eventIdBefore)) { + LOGGER.info("Successfully fetched hive full snapshot, Current NotificationID: {}.", + eventIdAfter); + // As eventIDAfter is the last event that was processed, eventIDAfter is used to update + // lastProcessedNotificationID instead of getting it from persistent store. + return new PathsImage(pathsFullSnapshot, eventIdAfter.getEventId(), + SentryStore.EMPTY_PATHS_SNAPSHOT_ID); + } + + LOGGER.info("Reconciling full snapshot - applying {} changes", + eventIdAfter.getEventId() - eventIdBefore.getEventId()); + + // While we were taking snapshot, HMS made some changes, so now we need to apply all + // extra events to the snapshot + long currentEventId = eventIdBefore.getEventId(); + SentryJSONMessageDeserializer deserializer = new SentryJSONMessageDeserializer(); + + while (currentEventId < eventIdAfter.getEventId()) { + NotificationEventResponse response = + client.getNextNotification(currentEventId, Integer.MAX_VALUE, null); + if (response == null || !response.isSetEvents() || response.getEvents().isEmpty()) { + LOGGER.error("Snapshot discarded, updates to HMS data while shapshot is being taken." + + "ID Before: {}. ID After: {}", eventIdBefore.getEventId(), eventIdAfter.getEventId()); + return new PathsImage(Collections.<String, Set<String>>emptyMap(), + SentryStore.EMPTY_NOTIFICATION_ID, SentryStore.EMPTY_PATHS_SNAPSHOT_ID); + } + + for (NotificationEvent event : response.getEvents()) { + if (event.getEventId() <= eventIdBefore.getEventId()) { + LOGGER.error("Received stray event with eventId {} which is less then {}", + event.getEventId(), eventIdBefore); + continue; + } + if (event.getEventId() > eventIdAfter.getEventId()) { + // Enough events processed + break; + } + try { + FullUpdateModifier.applyEvent(pathsFullSnapshot, event, deserializer); + } catch (Exception e) { + LOGGER.warn("Failed to apply operation", e); + } + currentEventId = event.getEventId(); + } } LOGGER.info("Successfully fetched hive full snapshot, Current NotificationID: {}.", - eventIdAfter); + currentEventId); // As eventIDAfter is the last event that was processed, eventIDAfter is used to update // lastProcessedNotificationID instead of getting it from persistent store. - return new PathsImage(pathsFullSnapshot, eventIdAfter.getEventId(), + return new PathsImage(pathsFullSnapshot, currentEventId, SentryStore.EMPTY_PATHS_SNAPSHOT_ID); } catch (TException failure) { LOGGER.error("Failed to communicate to HMS"); @@ -198,7 +238,7 @@ class SentryHMSClient implements AutoCloseable { } catch (Exception ignored) { failedSnapshotsCount.inc(); LOGGER.error("Snapshot created failed ", ignored); - return Collections.emptyMap(); + return emptyMap(); } } @@ -210,7 +250,7 @@ class SentryHMSClient implements AutoCloseable { */ Collection<NotificationEvent> getNotifications(long notificationId) throws Exception { if (client == null) { - LOGGER.error("Client is not connected to HMS"); + LOGGER.error(NOT_CONNECTED_MSG); return Collections.emptyList(); } @@ -246,9 +286,9 @@ class SentryHMSClient implements AutoCloseable { * @return the latest notification Id logged by the HMS * @throws Exception when an error occurs when talking to the HMS client */ - public long getCurrentNotificationId() throws Exception { + long getCurrentNotificationId() throws Exception { if (client == null) { - LOGGER.error("Client is not connected to HMS"); + LOGGER.error(NOT_CONNECTED_MSG); return SentryStore.EMPTY_NOTIFICATION_ID; } http://git-wip-us.apache.org/repos/asf/sentry/blob/10e65dad/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/service/thrift/TestFullUpdateModifier.java ---------------------------------------------------------------------- diff --git a/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/service/thrift/TestFullUpdateModifier.java b/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/service/thrift/TestFullUpdateModifier.java new file mode 100644 index 0000000..7deccb0 --- /dev/null +++ b/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/service/thrift/TestFullUpdateModifier.java @@ -0,0 +1,453 @@ +/** + * 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.sentry.service.thrift; + +import org.apache.hadoop.hive.metastore.api.NotificationEvent; +import org.apache.hive.hcatalog.messaging.MessageDeserializer; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONAddPartitionMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONAlterPartitionMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONAlterTableMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONCreateDatabaseMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONCreateTableMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONDropDatabaseMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONDropPartitionMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONDropTableMessage; +import org.apache.sentry.binding.metastore.messaging.json.SentryJSONMessageDeserializer; +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +import static org.apache.hive.hcatalog.messaging.HCatEventMessage.EventType.*; +import static org.junit.Assert.*; + +public class TestFullUpdateModifier { + private static final String SERVER = "s"; + private static final String PRINCIPAL = "p"; + private static final String DB = "Db1"; + private static final String TABLE = "Tab1"; + private static final String AUTH = DB.toLowerCase() + "." + TABLE.toLowerCase(); + private static final String PATH = "foo/bar"; + private static final String LOCATION = uri(PATH); + + /** + * Convert path to HDFS URI + */ + private static final String uri(String path) { + return "hdfs:///" + path; + } + + /** + * Test create database event. It should add database and its location. + * As a result we should have entry {"db1": {foo/bar}} + * @throws Exception + */ + @Test + public void testCreateDatabase() throws Exception { + Map<String, Set<String>> update = new HashMap<>(); + NotificationEvent event = new NotificationEvent(0, 0, CREATE_DATABASE.toString(), ""); + MessageDeserializer deserializer = Mockito.mock(SentryJSONMessageDeserializer.class); + + SentryJSONCreateDatabaseMessage message = + new SentryJSONCreateDatabaseMessage(SERVER, PRINCIPAL, DB, 0L, LOCATION); + Mockito.when(deserializer.getCreateDatabaseMessage("")).thenReturn(message); + FullUpdateModifier.applyEvent(update, event, deserializer); + Map<String, Set<String>> expected = new HashMap<>(); + expected.put(DB.toLowerCase(), Collections.singleton(PATH)); + assertEquals(expected, update); + } + + /** + * Test drop database event. It should drop database record. + * @throws Exception + */ + @Test + public void testDropDatabase() throws Exception { + Map<String, Set<String>> update = new HashMap<>(); + update.put(DB.toLowerCase(), Collections.singleton(PATH)); + NotificationEvent event = new NotificationEvent(0, 0, DROP_DATABASE.toString(), ""); + MessageDeserializer deserializer = Mockito.mock(SentryJSONMessageDeserializer.class); + + SentryJSONDropDatabaseMessage message = + new SentryJSONDropDatabaseMessage(SERVER, PRINCIPAL, DB, 0L, LOCATION); + Mockito.when(deserializer.getDropDatabaseMessage("")).thenReturn(message); + FullUpdateModifier.applyEvent(update, event, deserializer); + assertTrue(update.isEmpty()); + } + + /** + * Test drop database event when dropped database location doesn't + * match original database location. Should leave update intact. + * @throws Exception + */ + @Test + public void testDropDatabaseWrongLocation() throws Exception { + Map<String, Set<String>> update = new HashMap<>(); + update.put(DB.toLowerCase(), Collections.singleton(PATH)); + + NotificationEvent event = new NotificationEvent(0, 0, DROP_DATABASE.toString(), ""); + MessageDeserializer deserializer = Mockito.mock(SentryJSONMessageDeserializer.class); + + SentryJSONDropDatabaseMessage message = + new SentryJSONDropDatabaseMessage(SERVER, PRINCIPAL, DB, 0L, + "hdfs:///bad/location"); + Mockito.when(deserializer.getDropDatabaseMessage("")).thenReturn(message); + FullUpdateModifier.applyEvent(update, event, deserializer); + // DB should stay + Map<String, Set<String>> expected = new HashMap<>(); + expected.put(DB.toLowerCase(), Collections.singleton(PATH)); + assertEquals(expected, update); + } + + /** + * Test drop database which has tables/partitions. + * Should drop all reated database records but leave unrelated records in place. + * @throws Exception + */ + @Test + public void testDropDatabaseWithTables() throws Exception { + Map<String, Set<String>> update = new HashMap<>(); + update.put(DB.toLowerCase(), Collections.singleton(PATH)); + update.put(AUTH, Collections.singleton(PATH)); + update.put("unrelated", Collections.singleton(PATH)); + NotificationEvent event = new NotificationEvent(0, 0, DROP_DATABASE.toString(), ""); + MessageDeserializer deserializer = Mockito.mock(SentryJSONMessageDeserializer.class); + + SentryJSONDropDatabaseMessage message = + new SentryJSONDropDatabaseMessage(SERVER, PRINCIPAL, DB, 0L, LOCATION); + Mockito.when(deserializer.getDropDatabaseMessage("")).thenReturn(message); + FullUpdateModifier.applyEvent(update, event, deserializer); + Map<String, Set<String>> expected = new HashMap<>(); + expected.put("unrelated", Collections.singleton(PATH)); + assertEquals(expected, update); + } + + /** + * Test create table event. It should add table and its location. + * As a result we should have entry {"db1.tab1": {foo/bar}} + * @throws Exception + */ + @Test + public void testCreateTable() throws Exception { + Map<String, Set<String>> update = new HashMap<>(); + NotificationEvent event = new NotificationEvent(0, 0, CREATE_TABLE.toString(), ""); + MessageDeserializer deserializer = Mockito.mock(SentryJSONMessageDeserializer.class); + + SentryJSONCreateTableMessage message = + new SentryJSONCreateTableMessage(SERVER, PRINCIPAL, DB, TABLE, 0L, LOCATION); + Mockito.when(deserializer.getCreateTableMessage("")).thenReturn(message); + FullUpdateModifier.applyEvent(update, event, deserializer); + Map<String, Set<String>> expected = new HashMap<>(); + expected.put(AUTH, Collections.singleton(PATH)); + assertEquals(expected, update); + } + + /** + * Test drop table event. It should drop table record. + * @throws Exception + */ + @Test + public void testDropTable() throws Exception { + Map<String, Set<String>> update = new HashMap<>(); + update.put(AUTH, Collections.singleton(PATH)); + NotificationEvent event = new NotificationEvent(0, 0, DROP_TABLE.toString(), ""); + MessageDeserializer deserializer = Mockito.mock(SentryJSONMessageDeserializer.class); + + SentryJSONDropTableMessage message = + new SentryJSONDropTableMessage(SERVER, PRINCIPAL, DB, TABLE, 0L, LOCATION); + Mockito.when(deserializer.getDropTableMessage("")).thenReturn(message); + FullUpdateModifier.applyEvent(update, event, deserializer); + assertTrue(update.isEmpty()); + } + + /** + * Test drop table event. It should drop table record. + * @throws Exception + */ + @Test + public void testDropTableWrongLocation() throws Exception { + Map<String, Set<String>> update = new HashMap<>(); + update.put(AUTH, Collections.singleton(PATH)); + NotificationEvent event = new NotificationEvent(0, 0, DROP_TABLE.toString(), ""); + MessageDeserializer deserializer = Mockito.mock(SentryJSONMessageDeserializer.class); + + SentryJSONDropTableMessage message = + new SentryJSONDropTableMessage(SERVER, PRINCIPAL, DB, TABLE, 0L, + "hdfs:///bad/location"); + Mockito.when(deserializer.getDropTableMessage("")).thenReturn(message); + FullUpdateModifier.applyEvent(update, event, deserializer); + // DB should stay + assertEquals(Collections.singleton(PATH), update.get(AUTH)); + assertEquals(1, update.size()); + } + + /** + * Test add partition event. It should add table and its location. + * As a result we should have entry {"db1.tab1": {foo/bar, hello/world}} + * @throws Exception + */ + @Test + public void testAddPartition() throws Exception { + Map<String, Set<String>> update = new HashMap<>(); + Set<String> locations = new HashSet<>(); + locations.add(PATH); + update.put(AUTH, locations); + + NotificationEvent event = new NotificationEvent(0, 0, ADD_PARTITION.toString(), ""); + MessageDeserializer deserializer = Mockito.mock(SentryJSONMessageDeserializer.class); + + String partPath = "hello/world"; + String partLocation = uri(partPath); + + SentryJSONAddPartitionMessage message = + new SentryJSONAddPartitionMessage(SERVER, PRINCIPAL, DB, TABLE, + Collections.<Map<String,String>>emptyList(), 0L, + Collections.singletonList(partLocation)); + Mockito.when(deserializer.getAddPartitionMessage("")).thenReturn(message); + FullUpdateModifier.applyEvent(update, event, deserializer); + Set<String> expected = new HashSet<>(2); + expected.add(PATH); + expected.add(partPath); + assertEquals(expected, update.get(AUTH)); + } + + /** + * Test drop partition event. It should drop partition info from the list of locations. + * @throws Exception + */ + @Test + public void testDropPartitions() throws Exception { + String partPath = "hello/world"; + String partLocation = uri(partPath); + Map<String, Set<String>> update = new HashMap<>(); + Set<String> locations = new HashSet<>(); + locations.add(PATH); + locations.add(partPath); + update.put(AUTH, locations); + + NotificationEvent event = new NotificationEvent(0, 0, DROP_PARTITION.toString(), ""); + MessageDeserializer deserializer = Mockito.mock(SentryJSONMessageDeserializer.class); + + SentryJSONDropPartitionMessage message = + new SentryJSONDropPartitionMessage(SERVER, PRINCIPAL, DB, TABLE, + Collections.<Map<String,String>>emptyList(), 0L, Collections.singletonList(partLocation)); + Mockito.when(deserializer.getDropPartitionMessage("")).thenReturn(message); + FullUpdateModifier.applyEvent(update, event, deserializer); + assertEquals(Collections.singleton(PATH), update.get(AUTH)); + } + + /** + * Test alter partition event. It should change partition location + * @throws Exception + */ + @Test + public void testAlterPartition() throws Exception { + String partPath = "hello/world"; + String partLocation = uri(partPath); + + String newPath = "better/world"; + String newLocation = uri(newPath); + + Map<String, Set<String>> update = new HashMap<>(); + Set<String> locations = new HashSet<>(); + locations.add(PATH); + locations.add(partPath); + update.put(AUTH, locations); + + NotificationEvent event = new NotificationEvent(0, 0, ALTER_PARTITION.toString(), ""); + MessageDeserializer deserializer = Mockito.mock(SentryJSONMessageDeserializer.class); + + SentryJSONAlterPartitionMessage message = + new SentryJSONAlterPartitionMessage(SERVER, PRINCIPAL, DB, TABLE, + Collections.<String>emptyList(), Collections.<String>emptyList(), 0L, + partLocation, newLocation); + + Mockito.when(deserializer.getAlterPartitionMessage("")).thenReturn(message); + FullUpdateModifier.applyEvent(update, event, deserializer); + + Set<String> expected = new HashSet<>(2); + expected.add(PATH); + expected.add(newPath); + assertEquals(expected, update.get(AUTH)); + } + + /** + * Test alter table event that changes database name when there are no tables. + * @throws Exception + */ + @Test + public void testAlterTableChangeDbNameNoTables() throws Exception { + Map<String, Set<String>> update = new HashMap<>(); + update.put(DB.toLowerCase(), Collections.singleton(PATH)); + String newDbName = "Db2"; + + NotificationEvent event = new NotificationEvent(0, 0, ALTER_TABLE.toString(), ""); + event.setDbName(newDbName); + event.setTableName(TABLE); + + MessageDeserializer deserializer = Mockito.mock(SentryJSONMessageDeserializer.class); + + SentryJSONAlterTableMessage message = + new SentryJSONAlterTableMessage(SERVER, PRINCIPAL, DB, TABLE, 0L, + LOCATION, LOCATION); + + Mockito.when(deserializer.getAlterTableMessage("")).thenReturn(message); + FullUpdateModifier.applyEvent(update, event, deserializer); + assertEquals(Collections.singleton(PATH), update.get(newDbName.toLowerCase())); + assertFalse(update.containsKey(DB.toLowerCase())); + } + + @Test + /** + * Test alter table event that changes database name when there are tables. + * All entries like "dbName.tableName" should have dbName changed to the new name. + * @throws Exception + */ + public void testAlterTableChangeDbNameWithTables() throws Exception { + Map<String, Set<String>> update = new HashMap<>(); + update.put(DB.toLowerCase(), Collections.singleton(PATH)); + Set<String> locations = new HashSet<>(1); + locations.add(PATH); + update.put(AUTH, locations); + + String newDbName = "Db2"; + String newAuth = newDbName.toLowerCase() + "." + TABLE.toLowerCase(); + + NotificationEvent event = new NotificationEvent(0, 0, ALTER_TABLE.toString(), ""); + event.setDbName(newDbName); + event.setTableName(TABLE); + + MessageDeserializer deserializer = Mockito.mock(SentryJSONMessageDeserializer.class); + + SentryJSONAlterTableMessage message = + new SentryJSONAlterTableMessage(SERVER, PRINCIPAL, DB, TABLE, 0L, + LOCATION, LOCATION); + + Mockito.when(deserializer.getAlterTableMessage("")).thenReturn(message); + FullUpdateModifier.applyEvent(update, event, deserializer); + Map<String, Set<String>> expected = new HashMap<>(2); + expected.put(newDbName.toLowerCase(), Collections.singleton(PATH)); + expected.put(newAuth, Collections.singleton(PATH)); + assertEquals(expected, update); + } + + /** + * Test alter table event that changes table name. + * @throws Exception + */ + @Test + public void testAlterTableChangeTableName() throws Exception { + Map<String, Set<String>> update = new HashMap<>(); + update.put(DB.toLowerCase(), Collections.singleton(PATH)); + Set<String> locations = new HashSet<>(1); + locations.add(PATH); + update.put(AUTH, locations); + + String newTableName = "Table2"; + String newAuth = DB.toLowerCase() + "." + newTableName.toLowerCase(); + + NotificationEvent event = new NotificationEvent(0, 0, ALTER_TABLE.toString(), ""); + event.setDbName(DB); + event.setTableName(newTableName); + + MessageDeserializer deserializer = Mockito.mock(SentryJSONMessageDeserializer.class); + + SentryJSONAlterTableMessage message = + new SentryJSONAlterTableMessage(SERVER, PRINCIPAL, DB, TABLE, 0L, + LOCATION, LOCATION); + + Mockito.when(deserializer.getAlterTableMessage("")).thenReturn(message); + FullUpdateModifier.applyEvent(update, event, deserializer); + Map<String, Set<String>> expected = new HashMap<>(2); + expected.put(DB.toLowerCase(), Collections.singleton(PATH)); + expected.put(newAuth, Collections.singleton(PATH)); + assertEquals(expected, update); + } + + /** + * Test alter table event that changes object location. + * @throws Exception + */ + @Test + public void testAlterTableChangeLocation() throws Exception { + Map<String, Set<String>> update = new HashMap<>(); + update.put(DB.toLowerCase(), Collections.singleton(PATH)); + Set<String> locations = new HashSet<>(1); + locations.add(PATH); + update.put(AUTH, locations); + + NotificationEvent event = new NotificationEvent(0, 0, ALTER_TABLE.toString(), ""); + event.setDbName(DB); + event.setTableName(TABLE); + + String newPath = "hello/world"; + String newLocation = uri(newPath); + + MessageDeserializer deserializer = Mockito.mock(SentryJSONMessageDeserializer.class); + + SentryJSONAlterTableMessage message = + new SentryJSONAlterTableMessage(SERVER, PRINCIPAL, DB, TABLE, 0L, + LOCATION, newLocation); + + Mockito.when(deserializer.getAlterTableMessage("")).thenReturn(message); + FullUpdateModifier.applyEvent(update, event, deserializer); + Map<String, Set<String>> expected = new HashMap<>(2); + expected.put(DB.toLowerCase(), Collections.singleton(PATH)); + expected.put(AUTH.toLowerCase(), Collections.singleton(newPath)); + assertEquals(expected, update); + } + + /** + * Test renamePrefixKeys function. + * We ask to rename "foo.bar" key to "foo.baz" key. + * @throws Exception + */ + @Test + public void testRenamePrefixKeys() throws Exception { + String oldKey = "foo."; + String newKey = "baz."; + String postfix = "bar"; + Map<String, Set<String>> update = new HashMap<>(); + update.put(oldKey + postfix , Collections.<String>emptySet()); + FullUpdateModifier.renamePrefixKeys(update, oldKey, newKey); + assertEquals(1, update.size()); + assertTrue(update.containsKey(newKey + postfix)); + } + + /** + * Test renamePostfixKeys and RenamePrefixKeys functions mwhen the destination keys exist. + * Should nto change anything. + * We ask to rename "foo.bar" key to "baz.bar" key. + * @throws Exception + */ + @Test + public void testRenameKeysWithConflicts() throws Exception { + Map<String, Set<String>> update = new HashMap<>(); + update.put("foo.bar", Collections.<String>emptySet()); + update.put("baz.bar", Collections.<String>emptySet()); + Map<String, Set<String>> expected = new HashMap<>(update); + + FullUpdateModifier.renamePrefixKeys(update, "foo.", "baz."); + assertEquals(update, expected); + } +} \ No newline at end of file
