You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sentry.apache.org by ak...@apache.org on 2017/07/10 23:02:57 UTC
sentry git commit: SENTRY-1815: Send new HMS snapshots to HDFS
requesting an old generation ID (Sergio Pena,
reviewed by Alex Kolbasov and Na Li)
Repository: sentry
Updated Branches:
refs/heads/sentry-ha-redesign c8d0f24a2 -> b9eca2144
SENTRY-1815: Send new HMS snapshots to HDFS requesting an old generation ID (Sergio Pena, reviewed by Alex Kolbasov and Na Li)
Project: http://git-wip-us.apache.org/repos/asf/sentry/repo
Commit: http://git-wip-us.apache.org/repos/asf/sentry/commit/b9eca214
Tree: http://git-wip-us.apache.org/repos/asf/sentry/tree/b9eca214
Diff: http://git-wip-us.apache.org/repos/asf/sentry/diff/b9eca214
Branch: refs/heads/sentry-ha-redesign
Commit: b9eca2144cb78062294f995a36f5f56d4c516ca5
Parents: c8d0f24
Author: Alexander Kolbasov <ak...@cloudera.com>
Authored: Tue Jul 11 01:02:34 2017 +0200
Committer: Alexander Kolbasov <ak...@cloudera.com>
Committed: Tue Jul 11 01:02:34 2017 +0200
----------------------------------------------------------------------
.../org/apache/sentry/hdfs/ImageRetriever.java | 5 +
.../org/apache/sentry/hdfs/PathsUpdate.java | 15 +-
.../apache/sentry/hdfs/PermissionsUpdate.java | 6 +
.../apache/sentry/hdfs/ServiceConstants.java | 9 ++
.../java/org/apache/sentry/hdfs/Updateable.java | 17 +++
.../sentry/hdfs/UpdateableAuthzPaths.java | 22 ++-
.../sentry/hdfs/SentryAuthorizationInfo.java | 40 ++---
.../org/apache/sentry/hdfs/SentryUpdater.java | 3 +-
.../sentry/hdfs/UpdateableAuthzPermissions.java | 15 +-
.../apache/sentry/hdfs/DBUpdateForwarder.java | 64 ++++++--
.../apache/sentry/hdfs/PathImageRetriever.java | 10 +-
.../apache/sentry/hdfs/PermImageRetriever.java | 5 +
.../sentry/hdfs/SentryHDFSServiceClient.java | 13 ++
.../SentryHDFSServiceClientDefaultImpl.java | 10 +-
.../sentry/hdfs/SentryHDFSServiceProcessor.java | 19 +--
.../org/apache/sentry/hdfs/SentryPlugin.java | 89 ++++++-----
.../sentry/hdfs/TestDBUpdateForwarder.java | 143 ++++++++++++++++++
.../apache/sentry/hdfs/TestDeltaRetriever.java | 74 ++++++++++
.../apache/sentry/hdfs/TestImageRetriever.java | 99 +++++++++++++
.../hdfs/TestSentryHDFSServiceProcessor.java | 146 +++++++++++++++++++
.../db/service/persistent/PathsImage.java | 12 +-
.../db/service/persistent/PermissionsImage.java | 4 +-
.../db/service/persistent/SentryStore.java | 23 ++-
.../db/service/persistent/TestSentryStore.java | 7 +-
24 files changed, 740 insertions(+), 110 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/ImageRetriever.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/ImageRetriever.java b/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/ImageRetriever.java
index e96140d..11b7541 100644
--- a/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/ImageRetriever.java
+++ b/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/ImageRetriever.java
@@ -41,4 +41,9 @@ public interface ImageRetriever<K extends Update> {
*/
K retrieveFullImage() throws Exception;
+ /**
+ * @return the latest image ID.
+ * @throws Exception if an error occurred requesting the image ID from the persistent storage.
+ */
+ long getLatestImageID() throws Exception;
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/PathsUpdate.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/PathsUpdate.java b/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/PathsUpdate.java
index 49befee..719c1ac 100644
--- a/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/PathsUpdate.java
+++ b/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/PathsUpdate.java
@@ -34,6 +34,8 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.thrift.TException;
+import static org.apache.sentry.hdfs.service.thrift.sentry_hdfs_serviceConstants.UNUSED_PATH_UPDATE_IMG_NUM;
+
/**
* A wrapper class over the TPathsUpdate thrift generated class. Please see
* {@link Updateable.Update} for more information
@@ -58,8 +60,12 @@ public class PathsUpdate implements Updateable.Update {
}
public PathsUpdate(long seqNum, boolean hasFullImage) {
- tPathsUpdate = new TPathsUpdate(hasFullImage, seqNum,
- new ArrayList<TPathChanges>());
+ this(seqNum, UNUSED_PATH_UPDATE_IMG_NUM, hasFullImage);
+ }
+
+ public PathsUpdate(long seqNum, long imgNum, boolean hasFullImage) {
+ tPathsUpdate = new TPathsUpdate(hasFullImage, seqNum, new ArrayList<TPathChanges>());
+ tPathsUpdate.setImgNum(imgNum);
}
@Override
@@ -89,6 +95,11 @@ public class PathsUpdate implements Updateable.Update {
tPathsUpdate.setSeqNum(seqNum);
}
+ @Override
+ public long getImgNum() {
+ return tPathsUpdate.getImgNum();
+ }
+
TPathsUpdate toThrift() {
return tPathsUpdate;
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/PermissionsUpdate.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/PermissionsUpdate.java b/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/PermissionsUpdate.java
index ebb0b96..7aa60a3 100644
--- a/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/PermissionsUpdate.java
+++ b/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/PermissionsUpdate.java
@@ -25,6 +25,7 @@ import java.util.HashMap;
import org.apache.sentry.hdfs.service.thrift.TPermissionsUpdate;
import org.apache.sentry.hdfs.service.thrift.TPrivilegeChanges;
import org.apache.sentry.hdfs.service.thrift.TRoleChanges;
+import org.apache.sentry.hdfs.service.thrift.sentry_hdfs_serviceConstants;
import org.apache.thrift.TException;
public class PermissionsUpdate implements Updateable.Update {
@@ -62,6 +63,11 @@ public class PermissionsUpdate implements Updateable.Update {
}
@Override
+ public long getImgNum() {
+ return sentry_hdfs_serviceConstants.UNUSED_PATH_UPDATE_IMG_NUM;
+ }
+
+ @Override
public boolean hasFullImage() {
return tPermUpdate.isHasfullImage();
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/ServiceConstants.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/ServiceConstants.java b/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/ServiceConstants.java
index 0741ebc..dee4dc2 100644
--- a/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/ServiceConstants.java
+++ b/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/ServiceConstants.java
@@ -28,6 +28,15 @@ public class ServiceConstants {
private static final ImmutableMap<String, String> SASL_PROPERTIES;
+ // number used in authz paths and permissions to request initial syncs
+ public static final long SEQUENCE_NUMBER_UPDATE_UNINITIALIZED = -1L;
+
+ // number used in authz paths and permissions to request initial syncs
+ public static final long IMAGE_NUMBER_UPDATE_UNINITIALIZED = 0L;
+
+ // number used in authz paths and permissions that specifies an unused image number
+ public static final long IMAGE_NUMBER_UPDATE_UNUSED = -1L;
+
static {
Map<String, String> saslProps = new HashMap<String, String>();
saslProps.put(Sasl.SERVER_AUTH, "true");
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/Updateable.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/Updateable.java b/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/Updateable.java
index 12baaa4..e777e4b 100644
--- a/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/Updateable.java
+++ b/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/Updateable.java
@@ -38,6 +38,8 @@ public interface Updateable<K extends Updateable.Update> {
void setSeqNum(long seqNum);
+ long getImgNum();
+
byte[] serialize() throws IOException;
void deserialize(byte data[]) throws IOException;
@@ -80,12 +82,27 @@ public interface Updateable<K extends Updateable.Update> {
long getLastUpdatedSeqNum();
/**
+ * Return image number of Last Update
+ * @return
+ */
+ long getLastUpdatedImgNum();
+
+ /**
* Create and Full image update of the local data structure
* @param currSeqNum
* @return
*/
+ @Deprecated
K createFullImageUpdate(long currSeqNum) throws Exception;
+ /**
+ * Create and Full image update of the local data structure
+ * @param currSeqNum
+ * @param currImgNum
+ * @return
+ */
+ K createFullImageUpdate(long currSeqNum, long currImgNum) throws Exception;
+
String getUpdateableTypeName();
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/UpdateableAuthzPaths.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/UpdateableAuthzPaths.java b/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/UpdateableAuthzPaths.java
index ad7f8c9..08a3b3e 100644
--- a/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/UpdateableAuthzPaths.java
+++ b/sentry-hdfs/sentry-hdfs-common/src/main/java/org/apache/sentry/hdfs/UpdateableAuthzPaths.java
@@ -28,12 +28,16 @@ import org.apache.sentry.hdfs.service.thrift.TPathsDump;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.sentry.hdfs.ServiceConstants.IMAGE_NUMBER_UPDATE_UNINITIALIZED;
+import static org.apache.sentry.hdfs.ServiceConstants.SEQUENCE_NUMBER_UPDATE_UNINITIALIZED;
+
public class UpdateableAuthzPaths implements AuthzPaths, Updateable<PathsUpdate> {
private static final int MAX_UPDATES_PER_LOCK_USE = 99;
private static final String UPDATABLE_TYPE_NAME = "path_update";
private static final Logger LOG = LoggerFactory.getLogger(UpdateableAuthzPaths.class);
private volatile HMSPaths paths;
- private final AtomicLong seqNum = new AtomicLong(0);
+ private final AtomicLong seqNum = new AtomicLong(SEQUENCE_NUMBER_UPDATE_UNINITIALIZED);
+ private final AtomicLong imgNum = new AtomicLong(IMAGE_NUMBER_UPDATE_UNINITIALIZED);
public UpdateableAuthzPaths(String[] pathPrefixes) {
this.paths = new HMSPaths(pathPrefixes);
@@ -63,6 +67,7 @@ public class UpdateableAuthzPaths implements AuthzPaths, Updateable<PathsUpdate>
UpdateableAuthzPaths other = getPathsDump().initializeFromDump(
update.toThrift().getPathsDump());
other.seqNum.set(update.getSeqNum());
+ other.imgNum.set(update.getImgNum());
return other;
}
@@ -79,7 +84,8 @@ public class UpdateableAuthzPaths implements AuthzPaths, Updateable<PathsUpdate>
lock.writeLock().lock();
}
seqNum.set(update.getSeqNum());
- LOG.debug("##### Updated paths seq Num [" + seqNum.get() + "]");
+ imgNum.set(update.getImgNum());
+ LOG.debug("##### Updated paths seq Num [{}] img Num [{}]", seqNum.get(), imgNum.get());
}
} finally {
lock.writeLock().unlock();
@@ -145,8 +151,18 @@ public class UpdateableAuthzPaths implements AuthzPaths, Updateable<PathsUpdate>
}
@Override
+ public long getLastUpdatedImgNum() {
+ return imgNum.get();
+ }
+
+ @Override
public PathsUpdate createFullImageUpdate(long currSeqNum) {
- PathsUpdate pathsUpdate = new PathsUpdate(currSeqNum, true);
+ throw new UnsupportedOperationException("createFullImageUpdate(currSeqNum)");
+ }
+
+ @Override
+ public PathsUpdate createFullImageUpdate(long currSeqNum, long currImgNum) throws Exception {
+ PathsUpdate pathsUpdate = new PathsUpdate(currSeqNum, currImgNum, true);
pathsUpdate.toThrift().setPathsDump(getPathsDump().createPathsDump());
return pathsUpdate;
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryAuthorizationInfo.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryAuthorizationInfo.java b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryAuthorizationInfo.java
index 90ba721..680db7a 100644
--- a/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryAuthorizationInfo.java
+++ b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryAuthorizationInfo.java
@@ -131,18 +131,19 @@ public class SentryAuthorizationInfo implements Runnable {
updates.getPathUpdates(), authzPaths);
UpdateableAuthzPermissions newAuthzPerms = processUpdates(
updates.getPermUpdates(), authzPermissions);
- // If there were any FULL updates the returned instance would be
- // different
+
+ // processUpdates() should return different newAuthzPaths and newAuthzPerms object references
+ // if FULL updates were fetched from the Sentry server, otherwise, the same authzPaths and authzPermissions
+ // objects will be returned.
if (newAuthzPaths != authzPaths || newAuthzPerms != authzPermissions) {
lock.writeLock().lock();
try {
- LOG.debug("FULL Updated paths seq Num [old="
- + authzPaths.getLastUpdatedSeqNum() + "], [new="
- + newAuthzPaths.getLastUpdatedSeqNum() + "]");
+ LOG.debug(String.format("FULL Updated paths seq Num [old=%d], [new=%d] img Num [old=%d], [new=%d]",
+ authzPaths.getLastUpdatedSeqNum(), newAuthzPaths.getLastUpdatedSeqNum(),
+ authzPaths.getLastUpdatedImgNum(), newAuthzPaths.getLastUpdatedImgNum()));
authzPaths = newAuthzPaths;
- LOG.debug("FULL Updated perms seq Num [old="
- + authzPermissions.getLastUpdatedSeqNum() + "], [new="
- + newAuthzPerms.getLastUpdatedSeqNum() + "]");
+ LOG.debug(String.format("FULL Updated perms seq Num [old=%d], [new=%d]",
+ authzPermissions.getLastUpdatedSeqNum(), newAuthzPerms.getLastUpdatedSeqNum()));
authzPermissions = newAuthzPerms;
} finally {
lock.writeLock().unlock();
@@ -160,22 +161,25 @@ public class SentryAuthorizationInfo implements Runnable {
V newUpdateable = updateable;
if (!updates.isEmpty()) {
if (updates.get(0).hasFullImage()) {
- LOG.debug("Process Update : FULL IMAGE "
- + "[" + newUpdateable.getClass() + "]"
- + "[" + updates.get(0).getSeqNum() + "]");
+ LOG.debug(String.format("Process Update : FULL IMAGE [%s][%d][%d]",
+ newUpdateable.getClass().getName(),
+ updates.get(0).getSeqNum(),
+ updates.get(0).getImgNum()));
newUpdateable = (V)newUpdateable.updateFull(updates.remove(0));
}
// Any more elements ?
if (!updates.isEmpty()) {
- LOG.debug("Process Update : More updates.. "
- + "[" + newUpdateable.getClass() + "]"
- + "[" + newUpdateable.getLastUpdatedSeqNum() + "]"
- + "[" + updates.size() + "]");
+ LOG.debug(String.format("Process Update : More updates.. [%s][%d][%d][%d]",
+ newUpdateable.getClass().getName(),
+ newUpdateable.getLastUpdatedSeqNum(),
+ newUpdateable.getLastUpdatedImgNum(),
+ updates.size()));
newUpdateable.updatePartial(updates, lock);
}
- LOG.debug("Process Update : Finished updates.. "
- + "[" + newUpdateable.getClass() + "]"
- + "[" + newUpdateable.getLastUpdatedSeqNum() + "]");
+ LOG.debug(String.format("Process Update : Finished updates.. [%s][%d][%d]",
+ newUpdateable.getClass().getName(),
+ newUpdateable.getLastUpdatedSeqNum(),
+ newUpdateable.getLastUpdatedImgNum()));
}
return newUpdateable;
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryUpdater.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryUpdater.java b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryUpdater.java
index 49f39b1..6c78a2a 100644
--- a/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryUpdater.java
+++ b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryUpdater.java
@@ -48,7 +48,8 @@ class SentryUpdater {
try {
return sentryClient.getAllUpdatesFrom(
authzInfo.getAuthzPermissions().getLastUpdatedSeqNum() + 1,
- authzInfo.getAuthzPaths().getLastUpdatedSeqNum() + 1);
+ authzInfo.getAuthzPaths().getLastUpdatedSeqNum() + 1,
+ authzInfo.getAuthzPaths().getLastUpdatedImgNum());
} catch (Exception e) {
sentryClient = null;
LOG.error("Error receiving updates from Sentry", e);
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/UpdateableAuthzPermissions.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/UpdateableAuthzPermissions.java b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/UpdateableAuthzPermissions.java
index 0259f44..89a3297 100644
--- a/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/UpdateableAuthzPermissions.java
+++ b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/UpdateableAuthzPermissions.java
@@ -31,9 +31,12 @@ import org.apache.sentry.hdfs.SentryPermissions.PrivilegeInfo;
import org.apache.sentry.hdfs.SentryPermissions.RoleInfo;
import org.apache.sentry.hdfs.service.thrift.TPrivilegeChanges;
import org.apache.sentry.hdfs.service.thrift.TRoleChanges;
+import org.apache.sentry.hdfs.service.thrift.sentry_hdfs_serviceConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.sentry.hdfs.ServiceConstants.SEQUENCE_NUMBER_UPDATE_UNINITIALIZED;
+
public class UpdateableAuthzPermissions implements AuthzPermissions, Updateable<PermissionsUpdate> {
private static final ImmutableMap<String, FsAction> ACTION_MAPPING = ImmutableMap.<String, FsAction>builder()
.put("ALL", FsAction.ALL)
@@ -46,7 +49,7 @@ public class UpdateableAuthzPermissions implements AuthzPermissions, Updateable<
private static final String UPDATABLE_TYPE_NAME = "perm_authz_update";
private static final Logger LOG = LoggerFactory.getLogger(UpdateableAuthzPermissions.class);
private final SentryPermissions perms = new SentryPermissions();
- private final AtomicLong seqNum = new AtomicLong(0);
+ private final AtomicLong seqNum = new AtomicLong(SEQUENCE_NUMBER_UPDATE_UNINITIALIZED);
@Override
public List<AclEntry> getAcls(String authzObj) {
@@ -219,6 +222,11 @@ public class UpdateableAuthzPermissions implements AuthzPermissions, Updateable<
}
@Override
+ public long getLastUpdatedImgNum() {
+ return sentry_hdfs_serviceConstants.UNUSED_PATH_UPDATE_IMG_NUM;
+ }
+
+ @Override
public PermissionsUpdate createFullImageUpdate(long currSeqNum) {
// Using in-memory cache perms to create a full permission snapshot.
PermissionsUpdate retVal = new PermissionsUpdate(currSeqNum, true);
@@ -238,6 +246,11 @@ public class UpdateableAuthzPermissions implements AuthzPermissions, Updateable<
}
@Override
+ public PermissionsUpdate createFullImageUpdate(long currSeqNum, long currImgNum) throws Exception {
+ throw new UnsupportedOperationException("createFullImageUpdate(currSeqNum, currImgNum");
+ }
+
+ @Override
public String getUpdateableTypeName() {
return UPDATABLE_TYPE_NAME;
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/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 f4086fb..1ab4d6f 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
@@ -20,12 +20,14 @@ package org.apache.sentry.hdfs;
import java.util.Collections;
import java.util.List;
-import org.apache.sentry.provider.db.service.persistent.SentryStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.concurrent.ThreadSafe;
+import static org.apache.sentry.hdfs.ServiceConstants.IMAGE_NUMBER_UPDATE_UNINITIALIZED;
+import static org.apache.sentry.hdfs.ServiceConstants.SEQUENCE_NUMBER_UPDATE_UNINITIALIZED;
+
/**
* DBUpdateForwarder propagates a complete snapshot or delta update of either
* Sentry Permissions ({@code PermissionsUpdate}) or Sentry representation of
@@ -48,35 +50,69 @@ class DBUpdateForwarder<K extends Updateable.Update> {
}
/**
- * Retrieves all delta updates from the requested sequence number (inclusive) from
- * a persistent storage.
- * It first checks if there is such newer deltas exists in the persistent storage.
+ * Retrieves all delta updates from the requested sequence number (inclusive) or a single full
+ * update with its own sequence number from a persistent storage.
+ * <p>
+ * As part of the requested sequence number, an image number may also be used that identifies whether
+ * new full updates are persisted and need to be retrieved instead of delta updates.
+ * <p>
+ * It first checks if there is such image number exists and/or has newer images persisted.
+ * If a newer image is found, then it will return it as a new single full update.
+ * Otherwise. it checks if there is such newer deltas exists in the persistent storage.
* If there is, returns a list of delta updates.
- * Otherwise, a complete snapshot will be returned.
+ * Otherwise, an empty list is returned.
*
- * @param seqNum the requested sequence number
- * @return a list of delta updates, e.g. {@link PathsUpdate} or {@link PermissionsUpdate}
+ * @param imgNum the requested image number (>= 0).
+ * A value < 0 is identified as an unused value, and full updates would be returned
+ * only if the sequence number if <= 0.
+ * @param seqNum the requested sequence number.
+ * Values <= 0 will be recognized as full updates request (unless an image number is used).
+ * @return a list of full or delta updates (a full update is returned as a single-element list),
+ * e.g. {@link PathsUpdate} or {@link PermissionsUpdate}
*/
- List<K> getAllUpdatesFrom(long seqNum) throws Exception {
+ List<K> getAllUpdatesFrom(long seqNum, long imgNum) throws Exception {
+ LOGGER.debug("GetAllUpdatesFrom sequence number {}, image number {}", seqNum, imgNum);
+
+ // An imgNum >= 0 are valid values for image identifiers (0 means a full update is requested)
+ if (imgNum >= IMAGE_NUMBER_UPDATE_UNINITIALIZED) {
+ long curImgNum = imageRetriever.getLatestImageID();
+ LOGGER.debug("Current image number is {}", curImgNum);
+
+ if (curImgNum == IMAGE_NUMBER_UPDATE_UNINITIALIZED) {
+ // Sentry has not fetched a full HMS snapshot yet.
+ return Collections.emptyList();
+ } else if (curImgNum > imgNum) {
+ // In case a new HMS snapshot has been processed, then return a full paths image.
+ LOGGER.info("A newer full update is found with image number: ", curImgNum);
+ return Collections.singletonList(imageRetriever.retrieveFullImage());
+ }
+ }
+
+ /*
+ * If no new images are found, then continue with checking for delta updates
+ */
+
long curSeqNum = deltaRetriever.getLatestDeltaID();
- LOGGER.debug("GetAllUpdatesFrom sequence number {}, current sequence number is {}",
- seqNum, curSeqNum);
+ LOGGER.debug("Current sequence number is {}", curSeqNum);
+
if (seqNum > curSeqNum) {
- // No new deltas requested
+ // No new notifications were processed.
return Collections.emptyList();
}
// Checks if newer deltas exist in the persistent storage.
// If there are, return the list of delta updates.
- if ((seqNum != SentryStore.INIT_CHANGE_ID) &&
- deltaRetriever.isDeltaAvailable(seqNum)) {
+ 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);
return deltas;
}
}
- // Return the full snapshot
+ // If the sequence number is < 0 or the requested delta is not available, then we
+ // return a full update.
+ LOGGER.info("A full update is returned due to an unavailable sequence number: ", seqNum);
return Collections.singletonList(imageRetriever.retrieveFullImage());
}
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/PathImageRetriever.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/PathImageRetriever.java b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/PathImageRetriever.java
index de94743..2426b40 100644
--- a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/PathImageRetriever.java
+++ b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/PathImageRetriever.java
@@ -25,8 +25,6 @@ import org.apache.sentry.provider.db.service.persistent.PathsImage;
import org.apache.sentry.provider.db.service.persistent.SentryStore;
import javax.annotation.concurrent.ThreadSafe;
-import java.util.ArrayList;
-import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -58,13 +56,14 @@ public class PathImageRetriever implements ImageRetriever<PathsUpdate> {
// delta change the snapshot corresponds to.
PathsImage pathsImage = sentryStore.retrieveFullPathsImage();
long curSeqNum = pathsImage.getCurSeqNum();
+ long curImgNum = pathsImage.getCurImgNum();
Map<String, Set<String>> pathImage = pathsImage.getPathImage();
// Translates the complete Hive paths snapshot into a PathsUpdate.
// Adds all <hiveObj, paths> mapping to be included in this paths update.
// And label it with the latest delta change sequence number for consumer
// to be aware of the next delta change it should continue with.
- PathsUpdate pathsUpdate = new PathsUpdate(curSeqNum, true);
+ PathsUpdate pathsUpdate = new PathsUpdate(curSeqNum, curImgNum, true);
for (Map.Entry<String, Set<String>> pathEnt : pathImage.entrySet()) {
TPathChanges pathChange = pathsUpdate.newPathChange(pathEnt.getKey());
@@ -86,4 +85,9 @@ public class PathImageRetriever implements ImageRetriever<PathsUpdate> {
return pathsUpdate;
}
}
+
+ @Override
+ public long getLatestImageID() throws Exception {
+ return sentryStore.getLastProcessedImageID();
+ }
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/PermImageRetriever.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/PermImageRetriever.java b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/PermImageRetriever.java
index bad1099..53ce34f 100644
--- a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/PermImageRetriever.java
+++ b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/PermImageRetriever.java
@@ -21,6 +21,7 @@ import com.codahale.metrics.Timer.Context;
import org.apache.sentry.hdfs.service.thrift.TPermissionsUpdate;
import org.apache.sentry.hdfs.service.thrift.TPrivilegeChanges;
import org.apache.sentry.hdfs.service.thrift.TRoleChanges;
+import org.apache.sentry.hdfs.service.thrift.sentry_hdfs_serviceConstants;
import org.apache.sentry.provider.db.service.persistent.PermissionsImage;
import org.apache.sentry.provider.db.service.persistent.SentryStore;
@@ -92,4 +93,8 @@ public class PermImageRetriever implements ImageRetriever<PermissionsUpdate> {
}
}
+ @Override
+ public long getLatestImageID() throws Exception {
+ return sentry_hdfs_serviceConstants.UNUSED_PATH_UPDATE_IMG_NUM;
+ }
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceClient.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceClient.java b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceClient.java
index b6b78bc..09449b9 100644
--- a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceClient.java
+++ b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceClient.java
@@ -35,7 +35,20 @@ public interface SentryHDFSServiceClient extends AutoCloseable {
* @return List of permission and path changes which may include a full snapshot.
* @throws SentryHdfsServiceException if a connection exception happens
*/
+ @Deprecated
SentryAuthzUpdate getAllUpdatesFrom(long permSeqNum, long pathSeqNum)
throws SentryHdfsServiceException;
+
+ /**
+ * Get any permission and path updates accumulated since given sequence numbers.
+ * May return full update.
+ * @param permSeqNum Last sequence number for permissions update processed by the NameNode plugin
+ * @param pathSeqNum Last sequence number for paths update processed by the NameNode plugin
+ * @param pathImgNum Last image number for paths update processed by the NameNode plugin
+ * @return List of permission and path changes which may include a full snapshot.
+ * @throws SentryHdfsServiceException if a connection exception happens
+ */
+ SentryAuthzUpdate getAllUpdatesFrom(long permSeqNum, long pathSeqNum, long pathImgNum)
+ throws SentryHdfsServiceException;
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceClientDefaultImpl.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceClientDefaultImpl.java b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceClientDefaultImpl.java
index 86d0f62..30d8adf 100644
--- a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceClientDefaultImpl.java
+++ b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceClientDefaultImpl.java
@@ -87,11 +87,15 @@ public class SentryHDFSServiceClientDefaultImpl
}
@Override
- public SentryAuthzUpdate getAllUpdatesFrom(long permSeqNum, long pathSeqNum)
+ public SentryAuthzUpdate getAllUpdatesFrom(long permSeqNum, long pathSeqNum) throws SentryHdfsServiceException {
+ return getAllUpdatesFrom(permSeqNum, pathSeqNum, UNUSED_PATH_UPDATE_IMG_NUM);
+ }
+
+ @Override
+ public SentryAuthzUpdate getAllUpdatesFrom(long permSeqNum, long pathSeqNum, long pathImgNum)
throws SentryHdfsServiceException {
try {
- TAuthzUpdateRequest updateRequest = new TAuthzUpdateRequest(permSeqNum, pathSeqNum,
- UNUSED_PATH_UPDATE_IMG_NUM);
+ TAuthzUpdateRequest updateRequest = new TAuthzUpdateRequest(permSeqNum, pathSeqNum, pathImgNum);
TAuthzUpdateResponse sentryUpdates = client.get_authz_updates(updateRequest);
List<PathsUpdate> pathsUpdates = Collections.emptyList();
if (sentryUpdates.getAuthzPathUpdate() != null) {
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceProcessor.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceProcessor.java b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceProcessor.java
index 1c7061b..6221f3d 100644
--- a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceProcessor.java
+++ b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceProcessor.java
@@ -33,8 +33,6 @@ import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.sentry.hdfs.service.thrift.sentry_hdfs_serviceConstants.UNUSED_PATH_UPDATE_IMG_NUM;
-
/**
* Process requests from HDFS Name Node plugin.
* The only supported request is {@link #get_all_authz_updates_from(long, long)}.
@@ -44,7 +42,8 @@ public class SentryHDFSServiceProcessor implements SentryHDFSService.Iface {
@Override
public TAuthzUpdateResponse get_all_authz_updates_from(long permSeqNum, long pathSeqNum) throws TException {
- return get_authz_updates(new TAuthzUpdateRequest(permSeqNum, pathSeqNum, UNUSED_PATH_UPDATE_IMG_NUM));
+ throw new UnsupportedOperationException(
+ "get_all_authz_updates_from() is not supported due to known bugs. Use get_authz_updates() instead.");
}
@Override
@@ -65,13 +64,15 @@ public class SentryHDFSServiceProcessor implements SentryHDFSService.Iface {
SentryPlugin.instance.getAllPermsUpdatesFrom(request.getPermSeqNum());
SentryHdfsMetricsUtil.getPermUpdateHistogram.update(permUpdates.size());
List<PathsUpdate> pathUpdates =
- SentryPlugin.instance.getAllPathsUpdatesFrom(request.getPathSeqNum());
+ SentryPlugin.instance.getAllPathsUpdatesFrom(request.getPathSeqNum(), request.getPathImgNum());
SentryHdfsMetricsUtil.getPathUpdateHistogram.update(pathUpdates.size());
List<TPathsUpdate> retPathUpdates = new ArrayList<>(pathUpdates.size());
for (PathsUpdate update : pathUpdates) {
- LOGGER.debug("Sending PATH preUpdate seq [{}], [{}]",
- update.getSeqNum(), update.toThrift());
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Sending PATH preUpdate seq [{}], [{}]",
+ update.getSeqNum(), update.getImgNum());
+ }
retPathUpdates.add(update.toThrift());
}
retVal.setAuthzPathUpdate(retPathUpdates);
@@ -93,11 +94,11 @@ public class SentryHDFSServiceProcessor implements SentryHDFSService.Iface {
StringBuilder pathSeq = new StringBuilder("<");
for (PathsUpdate pathUpdate : pathUpdates) {
pathSeq.append(pathUpdate.getSeqNum()).append(",");
+ pathSeq.append(pathUpdate.getImgNum()).append(",");
}
pathSeq.append(">");
- LOGGER.debug("Updates requested from HDFS ["
- + "permReq=" + request.getPermSeqNum() + ", permResp=" + permSeq + "] "
- + "[pathReq=" + request.getPathSeqNum() + ", pathResp=" + pathSeq + "]");
+ LOGGER.debug("Updates requested from HDFS [permReq={}, permResp={}] [pathReq={}, pathResp={}]",
+ new Object[]{request.getPermSeqNum(), permSeq, request.getPathSeqNum(), pathSeq});
}
} catch (Exception e) {
LOGGER.error("Error Sending updates to downstream Cache", e);
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryPlugin.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryPlugin.java b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryPlugin.java
index 0bd0833..d6100de 100644
--- a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryPlugin.java
+++ b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryPlugin.java
@@ -21,7 +21,6 @@ package org.apache.sentry.hdfs;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
import org.apache.hadoop.conf.Configuration;
import org.apache.sentry.core.common.utils.SigUtils;
@@ -43,9 +42,12 @@ import org.apache.sentry.service.thrift.HMSFollower;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.sentry.hdfs.ServiceConstants.IMAGE_NUMBER_UPDATE_UNINITIALIZED;
+import static org.apache.sentry.hdfs.ServiceConstants.SEQUENCE_NUMBER_UPDATE_UNINITIALIZED;
import static org.apache.sentry.hdfs.Updateable.Update;
+import static org.apache.sentry.hdfs.service.thrift.sentry_hdfs_serviceConstants.UNUSED_PATH_UPDATE_IMG_NUM;
- /**
+/**
* SentryPlugin listens to all sentry permission update events, persists permission
* changes into database. It also facilitates HDFS synchronization between HMS and NameNode.
* <p>
@@ -60,8 +62,11 @@ import static org.apache.sentry.hdfs.Updateable.Update;
* or more updates previously received via HMS notification log.
* </ol>
* <p>
- * Each individual update is assigned a corresponding sequence number to synchronize
- * updates between Sentry and NameNode.
+ * Each individual update is assigned a corresponding sequence number and an image number
+ * to synchronize updates between Sentry and NameNode.
+ * <p>
+ * The image number may be used to identify whether new full updates are persisted and need
+ * to be retrieved instead of delta updates.
* <p>
* SentryPlugin also implements signal-triggered mechanism of full path
* updates from HMS to Sentry and from Sentry to NameNode, to address
@@ -94,15 +99,6 @@ public class SentryPlugin implements SentryPolicyStorePlugin, SigUtils.SigListen
private DBUpdateForwarder<PathsUpdate> pathsUpdater;
private DBUpdateForwarder<PermissionsUpdate> permsUpdater;
- /*
- * This number is smaller than starting sequence numbers used by NN and HMS
- * so in both cases its effect is to create appearance of out-of-sync
- * updates on the Sentry server (as if there were no previous updates at all).
- * It, in turn, triggers a) pushing full update from HMS to Sentry and
- * b) pulling full update from Sentry to NameNode.
- */
- private static final long NO_LAST_SEEN_HMS_PATH_SEQ_NUM = 0L;
-
@Override
public void initialize(Configuration conf, SentryStore sentryStore) throws SentryPluginException {
PermImageRetriever permImageRetriever = new PermImageRetriever(sentryStore);
@@ -133,49 +129,50 @@ public class SentryPlugin implements SentryPolicyStorePlugin, SigUtils.SigListen
* Request for update from NameNode.
* Full update to NameNode should happen only after full update from HMS.
*/
- public List<PathsUpdate> getAllPathsUpdatesFrom(long pathSeqNum) throws Exception {
+ public List<PathsUpdate> getAllPathsUpdatesFrom(long pathSeqNum, long pathImgNum) throws Exception {
if (!fullUpdateNN.get()) {
// Most common case - Sentry is NOT handling a full update.
- return pathsUpdater.getAllUpdatesFrom(pathSeqNum);
- } else {
- /*
- * Sentry is in the middle of signal-triggered full update.
- * It already got a full update from HMS
- */
- LOGGER.info("SIGNAL HANDLING: sending full update to NameNode");
- fullUpdateNN.set(false); // don't do full NN update till the next signal
- List<PathsUpdate> updates = pathsUpdater.getAllUpdatesFrom(NO_LAST_SEEN_HMS_PATH_SEQ_NUM);
- /*
- * This code branch is only called when Sentry is in the middle of a full update
- * (fullUpdateNN == true) and Sentry has already received full update from HMS
- * (fullUpdateHMSWait == false). It means that the local cache has a full update
- * from HMS.
- *
- * The full update list is expected to contain the last full update as the first
- * element, followed by zero or more subsequent partial updates.
- *
- * Returning NULL, empty, or partial update instead would be unexplainable, so
- * it should be logged.
- */
- if (updates != null) {
- if (!updates.isEmpty()) {
- if (updates.get(0).hasFullImage()) {
- LOGGER.info("SIGNAL HANDLING: Confirmed full update to NameNode");
- } else {
- LOGGER.warn("SIGNAL HANDLING: Sending partial instead of full update to NameNode (???)");
- }
+ return pathsUpdater.getAllUpdatesFrom(pathSeqNum, pathImgNum);
+ }
+
+ /*
+ * Sentry is in the middle of signal-triggered full update.
+ * It already got a full update from HMS
+ */
+ LOGGER.info("SIGNAL HANDLING: sending full update to NameNode");
+ fullUpdateNN.set(false); // don't do full NN update till the next signal
+ List<PathsUpdate> updates =
+ pathsUpdater.getAllUpdatesFrom(SEQUENCE_NUMBER_UPDATE_UNINITIALIZED, IMAGE_NUMBER_UPDATE_UNINITIALIZED);
+ /*
+ * This code branch is only called when Sentry is in the middle of a full update
+ * (fullUpdateNN == true) and Sentry has already received full update from HMS
+ * (fullUpdateHMSWait == false). It means that the local cache has a full update
+ * from HMS.
+ *
+ * The full update list is expected to contain the last full update as the first
+ * element, followed by zero or more subsequent partial updates.
+ *
+ * Returning NULL, empty, or partial update instead would be unexplainable, so
+ * it should be logged.
+ */
+ if (updates != null) {
+ if (!updates.isEmpty()) {
+ if (updates.get(0).hasFullImage()) {
+ LOGGER.info("SIGNAL HANDLING: Confirmed full update to NameNode");
} else {
- LOGGER.warn("SIGNAL HANDLING: Sending empty instead of full update to NameNode (???)");
+ LOGGER.warn("SIGNAL HANDLING: Sending partial instead of full update to NameNode (???)");
}
} else {
- LOGGER.warn("SIGNAL HANDLING: returned NULL instead of full update to NameNode (???)");
+ LOGGER.warn("SIGNAL HANDLING: Sending empty instead of full update to NameNode (???)");
}
- return updates;
+ } else {
+ LOGGER.warn("SIGNAL HANDLING: returned NULL instead of full update to NameNode (???)");
}
+ return updates;
}
public List<PermissionsUpdate> getAllPermsUpdatesFrom(long permSeqNum) throws Exception {
- return permsUpdater.getAllUpdatesFrom(permSeqNum);
+ return permsUpdater.getAllUpdatesFrom(permSeqNum, UNUSED_PATH_UPDATE_IMG_NUM);
}
@Override
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestDBUpdateForwarder.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestDBUpdateForwarder.java b/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestDBUpdateForwarder.java
new file mode 100644
index 0000000..8fbc100
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestDBUpdateForwarder.java
@@ -0,0 +1,143 @@
+/**
+ * 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
+ * <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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.sentry.hdfs;
+
+import org.apache.sentry.provider.db.service.persistent.SentryStore;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.sentry.hdfs.service.thrift.sentry_hdfs_serviceConstants.UNUSED_PATH_UPDATE_IMG_NUM;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class TestDBUpdateForwarder {
+ private ImageRetriever imageRetriever;
+ private DeltaRetriever deltaRetriever;
+ private DBUpdateForwarder updater;
+
+ @Before
+ public void setUp() {
+ imageRetriever = Mockito.mock(ImageRetriever.class);
+ deltaRetriever = Mockito.mock(DeltaRetriever.class);
+ updater = new DBUpdateForwarder<>(imageRetriever, deltaRetriever);
+ }
+
+ @Test
+ public void testEmptyListIsReturnedWhenImageNumIsZeroAndNoImagesArePersisted() throws Exception {
+ Mockito.when(imageRetriever.getLatestImageID()).thenReturn(SentryStore.EMPTY_PATHS_SNAPSHOT_ID);
+
+ List updates = updater.getAllUpdatesFrom(1, SentryStore.EMPTY_PATHS_SNAPSHOT_ID);
+ assertTrue(updates.isEmpty());
+ }
+
+ @Test
+ public void testEmptyListIsReturnedWhenImageIsUnusedAndNoDeltaChangesArePersisted() throws Exception {
+ Mockito.when(deltaRetriever.getLatestDeltaID()).thenReturn(SentryStore.EMPTY_NOTIFICATION_ID);
+
+ List updates = updater.getAllUpdatesFrom(1, UNUSED_PATH_UPDATE_IMG_NUM);
+ assertTrue(updates.isEmpty());
+ }
+
+ @Test
+ public void testFirstImageSyncIsReturnedWhenImageNumIsZero() throws Exception {
+ Mockito.when(imageRetriever.getLatestImageID()).thenReturn(1L);
+ Mockito.when(imageRetriever.retrieveFullImage())
+ .thenReturn(new PathsUpdate(1, 1, true));
+
+ List<PathsUpdate> updates = updater.getAllUpdatesFrom(0, SentryStore.EMPTY_PATHS_SNAPSHOT_ID);
+ assertEquals(1, updates.size());
+ assertEquals(1, updates.get(0).getSeqNum());
+ assertEquals(1, updates.get(0).getImgNum());
+ assertTrue(updates.get(0).hasFullImage());
+ }
+
+ @Test
+ public void testFirstImageSyncIsReturnedWhenImageNumIsUnusedButDeltasAreAvailable() throws Exception {
+ Mockito.when(deltaRetriever.getLatestDeltaID()).thenReturn(1L);
+ Mockito.when(imageRetriever.retrieveFullImage())
+ .thenReturn(new PathsUpdate(1, 1, true));
+
+ List<PathsUpdate> updates = updater.getAllUpdatesFrom(0, UNUSED_PATH_UPDATE_IMG_NUM);
+ assertEquals(1, updates.size());
+ assertEquals(1, updates.get(0).getSeqNum());
+ assertEquals(1, updates.get(0).getImgNum());
+ assertTrue(updates.get(0).hasFullImage());
+ }
+
+ @Test
+ public void testNewImageUpdateIsReturnedWhenNewImagesArePersisted() throws Exception {
+ Mockito.when(imageRetriever.getLatestImageID()).thenReturn(2L);
+ Mockito.when(imageRetriever.retrieveFullImage())
+ .thenReturn(new PathsUpdate(1, 2, true));
+
+ List<PathsUpdate> updates = updater.getAllUpdatesFrom(1, 1);
+ assertEquals(1, updates.size());
+ assertEquals(1, updates.get(0).getSeqNum());
+ assertEquals(2, updates.get(0).getImgNum());
+ assertTrue(updates.get(0).hasFullImage());
+ }
+
+ @Test
+ public void testNewImageUpdateIsReturnedWhenRequestedDeltaIsNotAvailable() throws Exception {
+ Mockito.when(imageRetriever.getLatestImageID()).thenReturn(1L);
+ Mockito.when(deltaRetriever.getLatestDeltaID()).thenReturn(3L);
+ Mockito.when(deltaRetriever.isDeltaAvailable(2L)).thenReturn(false);
+ Mockito.when(imageRetriever.retrieveFullImage())
+ .thenReturn(new PathsUpdate(3, 1, true));
+
+ List<PathsUpdate> updates = updater.getAllUpdatesFrom(2, 1);
+ assertEquals(1, updates.size());
+ assertEquals(3, updates.get(0).getSeqNum());
+ assertEquals(1, updates.get(0).getImgNum());
+ assertTrue(updates.get(0).hasFullImage());
+ }
+
+ @Test
+ public void testNewDeltasAreReturnedWhenRequestedDeltaIsAvailable() throws Exception {
+ Mockito.when(imageRetriever.getLatestImageID()).thenReturn(1L);
+ Mockito.when(deltaRetriever.getLatestDeltaID()).thenReturn(3L);
+ Mockito.when(deltaRetriever.isDeltaAvailable(2L)).thenReturn(true);
+ Mockito.when(deltaRetriever.retrieveDelta(2L))
+ .thenReturn(Arrays.asList(new PathsUpdate(3, 1, false)));
+
+ List<PathsUpdate> updates = updater.getAllUpdatesFrom(2, 1);
+ assertEquals(1, updates.size());
+ assertEquals(3, updates.get(0).getSeqNum());
+ assertEquals(1, updates.get(0).getImgNum());
+ assertFalse(updates.get(0).hasFullImage());
+ }
+
+ @Test
+ public void testNewImageIsReturnedWhenZeroSeqNumAndUnusedImgNumAreUsed() throws Exception {
+ Mockito.when(imageRetriever.getLatestImageID()).thenReturn(0L);
+ Mockito.when(deltaRetriever.getLatestDeltaID()).thenReturn(0L);
+ Mockito.when(imageRetriever.retrieveFullImage())
+ .thenReturn(new PermissionsUpdate(1, true));
+
+ List<PermissionsUpdate> updates = updater.getAllUpdatesFrom(0, UNUSED_PATH_UPDATE_IMG_NUM);
+ assertEquals(1, updates.size());
+ assertEquals(1, updates.get(0).getSeqNum());
+ assertEquals(UNUSED_PATH_UPDATE_IMG_NUM, updates.get(0).getImgNum());
+ assertTrue(updates.get(0).hasFullImage());
+ }
+}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestDeltaRetriever.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestDeltaRetriever.java b/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestDeltaRetriever.java
new file mode 100644
index 0000000..7ea75a0
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestDeltaRetriever.java
@@ -0,0 +1,74 @@
+/**
+ * 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.hdfs;
+
+import org.apache.sentry.provider.db.service.model.MSentryPathChange;
+import org.apache.sentry.provider.db.service.persistent.SentryStore;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class TestDeltaRetriever {
+ SentryStore sentryStoreMock;
+
+ @Before
+ public void setUp() {
+ sentryStoreMock = Mockito.mock(SentryStore.class);
+ }
+
+ @Test
+ public void testEmptyPathUpdatesRetrieveWhenNotPathChangesArePersisted() throws Exception {
+ Mockito.when(sentryStoreMock.getMSentryPathChanges(Mockito.anyLong()))
+ .thenReturn(Collections.<MSentryPathChange>emptyList());
+
+ PathDeltaRetriever deltaRetriever = new PathDeltaRetriever(sentryStoreMock);
+ List<PathsUpdate> pathsUpdates = deltaRetriever.retrieveDelta(1);
+
+ assertTrue(pathsUpdates.isEmpty());
+ }
+
+ @Test
+ public void testDeltaPathUpdatesRetrievedWhenNewPathChangesArePersisted() throws Exception {
+ PathDeltaRetriever deltaRetriever;
+ List<PathsUpdate> pathsUpdates;
+
+ List<MSentryPathChange> deltaPathChanges = Arrays.asList(
+ new MSentryPathChange(1, new PathsUpdate(1, true)),
+ new MSentryPathChange(2, new PathsUpdate(2, false))
+ );
+
+ Mockito.when(sentryStoreMock.getMSentryPathChanges(Mockito.anyLong()))
+ .thenReturn(deltaPathChanges);
+
+ deltaRetriever = new PathDeltaRetriever(sentryStoreMock);
+ pathsUpdates = deltaRetriever.retrieveDelta(1);
+
+ assertEquals(2, pathsUpdates.size());
+ assertEquals(1, pathsUpdates.get(0).getSeqNum());
+ assertEquals(true, pathsUpdates.get(0).hasFullImage());
+ assertEquals(2, pathsUpdates.get(1).getSeqNum());
+ assertEquals(false, pathsUpdates.get(1).hasFullImage());
+ }
+}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestImageRetriever.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestImageRetriever.java b/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestImageRetriever.java
new file mode 100644
index 0000000..20b3e10
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestImageRetriever.java
@@ -0,0 +1,99 @@
+/**
+ * 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.hdfs;
+
+import com.google.common.collect.Sets;
+import org.apache.commons.lang.StringUtils;
+import org.apache.sentry.hdfs.service.thrift.TPathChanges;
+import org.apache.sentry.provider.db.service.persistent.PathsImage;
+import org.apache.sentry.provider.db.service.persistent.SentryStore;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class TestImageRetriever {
+ SentryStore sentryStoreMock;
+
+ @Before
+ public void setUp() {
+ sentryStoreMock = Mockito.mock(SentryStore.class);
+ }
+
+ @Test
+ public void testEmptyPathUpdatesRetrievedWhenNoImagesArePersisted() throws Exception {
+ Mockito.when(sentryStoreMock.retrieveFullPathsImage())
+ .thenReturn(new PathsImage(new HashMap<String, Set<String>>(), 0, 0));
+
+ PathImageRetriever imageRetriever = new PathImageRetriever(sentryStoreMock);
+ PathsUpdate pathsUpdate = imageRetriever.retrieveFullImage();
+
+ assertEquals(0, pathsUpdate.getImgNum());
+ assertEquals(0, pathsUpdate.getSeqNum());
+ assertTrue(pathsUpdate.getPathChanges().isEmpty());
+ }
+
+ @Test
+ public void testFullPathUpdatesRetrievedWhenNewImagesArePersisted() throws Exception {
+ PathImageRetriever imageRetriever;
+ PathsUpdate pathsUpdate;
+
+ Map<String, Set<String>> fullPathsImage = new HashMap<>();
+ fullPathsImage.put("db1", Sets.newHashSet("/user/db1"));
+ fullPathsImage.put("db1.table1", Sets.newHashSet("/user/db1/table1"));
+
+ Mockito.when(sentryStoreMock.retrieveFullPathsImage())
+ .thenReturn(new PathsImage(fullPathsImage, 1, 1));
+
+ imageRetriever = new PathImageRetriever(sentryStoreMock);
+ pathsUpdate = imageRetriever.retrieveFullImage();
+
+ assertEquals(1, pathsUpdate.getImgNum());
+ assertEquals(1, pathsUpdate.getSeqNum());
+ assertEquals(2, pathsUpdate.getPathChanges().size());
+ assertTrue(comparePaths(fullPathsImage, pathsUpdate.getPathChanges()));
+ }
+
+ private boolean comparePaths(Map<String, Set<String>> expected, List<TPathChanges> actual) {
+ if (expected.size() != actual.size()) {
+ return false;
+ }
+
+ for (TPathChanges pathChanges : actual) {
+ if (!expected.containsKey(pathChanges.getAuthzObj())) {
+ return false;
+ }
+
+ Set<String> expectedPaths = expected.get(pathChanges.getAuthzObj());
+ for (List<String> path : pathChanges.getAddPaths()) {
+ if (!expectedPaths.contains(StringUtils.join(path, "/"))) {
+ return false;
+ }
+ }
+ }
+
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestSentryHDFSServiceProcessor.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestSentryHDFSServiceProcessor.java b/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestSentryHDFSServiceProcessor.java
new file mode 100644
index 0000000..f2368b7
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-service/src/test/java/org/apache/sentry/hdfs/TestSentryHDFSServiceProcessor.java
@@ -0,0 +1,146 @@
+/**
+ * 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
+ * <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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.sentry.hdfs;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.sentry.hdfs.service.thrift.TAuthzUpdateRequest;
+import org.apache.sentry.hdfs.service.thrift.TAuthzUpdateResponse;
+import org.apache.sentry.provider.db.SentryPolicyStorePlugin;
+import org.apache.sentry.provider.db.service.model.MSentryPathChange;
+import org.apache.sentry.provider.db.service.model.MSentryPermChange;
+import org.apache.sentry.provider.db.service.persistent.PathsImage;
+import org.apache.sentry.provider.db.service.persistent.PermissionsImage;
+import org.apache.sentry.provider.db.service.persistent.SentryStore;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.*;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class TestSentryHDFSServiceProcessor {
+ private static SentryHDFSServiceProcessor serviceProcessor;
+ private static SentryStore sentryStoreMock;
+
+ @BeforeClass
+ public static void setUp() throws SentryPolicyStorePlugin.SentryPluginException {
+ serviceProcessor = new SentryHDFSServiceProcessor();
+ sentryStoreMock = Mockito.mock(SentryStore.class);
+ new SentryPlugin().initialize(new Configuration(), sentryStoreMock);
+ }
+
+ @Test
+ public void testInitialHDFSSyncReturnsAFullImage() throws Exception {
+ Mockito.when(sentryStoreMock.getLastProcessedImageID())
+ .thenReturn(1L);
+ Mockito.when(sentryStoreMock.retrieveFullPathsImage())
+ .thenReturn(new PathsImage(new HashMap<String, Set<String>>(), 1, 1));
+
+ Mockito.when(sentryStoreMock.getLastProcessedPermChangeID())
+ .thenReturn(1L);
+ Mockito.when(sentryStoreMock.retrieveFullPermssionsImage())
+ .thenReturn(new PermissionsImage(new HashMap<String, List<String>>(), new HashMap<String, Map<String, String>>(), 1));
+
+ TAuthzUpdateRequest updateRequest = new TAuthzUpdateRequest(1, 1, 0);
+ TAuthzUpdateResponse sentryUpdates= serviceProcessor.get_authz_updates(updateRequest);
+
+ assertEquals(1, sentryUpdates.getAuthzPathUpdateSize());
+ assertEquals(1, sentryUpdates.getAuthzPathUpdate().get(0).getImgNum());
+ assertEquals(1, sentryUpdates.getAuthzPathUpdate().get(0).getSeqNum());
+ assertTrue(sentryUpdates.getAuthzPathUpdate().get(0).isHasFullImage());
+
+ assertEquals(1, sentryUpdates.getAuthzPermUpdateSize());
+ assertEquals(1, sentryUpdates.getAuthzPermUpdate().get(0).getSeqNum());
+ assertTrue(sentryUpdates.getAuthzPermUpdate().get(0).isHasfullImage());
+ }
+
+ @Test
+ public void testRequestSyncUpdatesWhenNewImagesArePersistedReturnsANewFullImage() throws Exception {
+ Mockito.when(sentryStoreMock.getLastProcessedImageID())
+ .thenReturn(2L);
+ Mockito.when(sentryStoreMock.retrieveFullPathsImage())
+ .thenReturn(new PathsImage(new HashMap<String, Set<String>>(), 3, 2));
+
+ Mockito.when(sentryStoreMock.getLastProcessedPermChangeID())
+ .thenReturn(3L);
+ Mockito.when(sentryStoreMock.retrieveFullPermssionsImage())
+ .thenReturn(new PermissionsImage(new HashMap<String, List<String>>(), new HashMap<String, Map<String, String>>(), 3));
+
+ TAuthzUpdateRequest updateRequest = new TAuthzUpdateRequest(2, 2, 1);
+ TAuthzUpdateResponse sentryUpdates= serviceProcessor.get_authz_updates(updateRequest);
+
+ assertEquals(1, sentryUpdates.getAuthzPathUpdateSize());
+ assertEquals(2, sentryUpdates.getAuthzPathUpdate().get(0).getImgNum());
+ assertEquals(3, sentryUpdates.getAuthzPathUpdate().get(0).getSeqNum());
+ assertTrue(sentryUpdates.getAuthzPathUpdate().get(0).isHasFullImage());
+
+ assertEquals(1, sentryUpdates.getAuthzPermUpdateSize());
+ assertEquals(3, sentryUpdates.getAuthzPermUpdate().get(0).getSeqNum());
+ assertTrue(sentryUpdates.getAuthzPermUpdate().get(0).isHasfullImage());
+ }
+
+ @Test
+ public void testRequestSyncUpdatesWhenNewDeltasArePersistedReturnsDeltaChanges() throws Exception {
+ Mockito.when(sentryStoreMock.getLastProcessedImageID())
+ .thenReturn(1L);
+ Mockito.when(sentryStoreMock.getLastProcessedPathChangeID())
+ .thenReturn(3L);
+ Mockito.when(sentryStoreMock.pathChangeExists(2))
+ .thenReturn(true);
+ Mockito.when(sentryStoreMock.getMSentryPathChanges(2))
+ .thenReturn(Arrays.asList(new MSentryPathChange(3, new PathsUpdate(3, 1, false))));
+
+ Mockito.when(sentryStoreMock.getLastProcessedPermChangeID())
+ .thenReturn(3L);
+ Mockito.when(sentryStoreMock.permChangeExists(2))
+ .thenReturn(true);
+ Mockito.when(sentryStoreMock.getMSentryPermChanges(2))
+ .thenReturn(Arrays.asList(new MSentryPermChange(3, new PermissionsUpdate(3, false))));
+
+ TAuthzUpdateRequest updateRequest = new TAuthzUpdateRequest(2, 2, 1);
+ TAuthzUpdateResponse sentryUpdates= serviceProcessor.get_authz_updates(updateRequest);
+
+ assertEquals(1, sentryUpdates.getAuthzPathUpdateSize());
+ assertEquals(1, sentryUpdates.getAuthzPathUpdate().get(0).getImgNum());
+ assertEquals(3, sentryUpdates.getAuthzPathUpdate().get(0).getSeqNum());
+ assertFalse(sentryUpdates.getAuthzPathUpdate().get(0).isHasFullImage());
+
+ assertEquals(1, sentryUpdates.getAuthzPermUpdateSize());
+ assertEquals(3, sentryUpdates.getAuthzPermUpdate().get(0).getSeqNum());
+ assertFalse(sentryUpdates.getAuthzPermUpdate().get(0).isHasfullImage());
+ }
+
+ @Test
+ public void testRequestSyncUpdatesWhenNoUpdatesExistReturnsEmptyResults() throws Exception {
+ Mockito.when(sentryStoreMock.getLastProcessedImageID())
+ .thenReturn(1L);
+ Mockito.when(sentryStoreMock.getLastProcessedPathChangeID())
+ .thenReturn(2L);
+ Mockito.when(sentryStoreMock.getLastProcessedPermChangeID())
+ .thenReturn(2L);
+
+ TAuthzUpdateRequest updateRequest = new TAuthzUpdateRequest(3, 3, 1);
+ TAuthzUpdateResponse sentryUpdates= serviceProcessor.get_authz_updates(updateRequest);
+
+ assertEquals(0, sentryUpdates.getAuthzPathUpdateSize());
+ assertEquals(0, sentryUpdates.getAuthzPermUpdateSize());
+ }
+}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/PathsImage.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/PathsImage.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/PathsImage.java
index fd56ce2..4d852e6 100644
--- a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/PathsImage.java
+++ b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/PathsImage.java
@@ -24,24 +24,30 @@ import java.util.Set;
/**
* A container for complete hive paths snapshot.
* <p>
- * It is composed by a hiveObj to Paths mapping and the sequence number/change ID
+ * It is composed by a hiveObj to Paths mapping, a paths image ID and the sequence number/change ID
* of latest delta change that the snapshot maps to.
*/
public class PathsImage {
- // A full snapshot of hiveObj to Paths mapping.
+ // A full image of hiveObj to Paths mapping.
private final Map<String, Set<String>> pathImage;
private final long curSeqNum;
+ private final long curImgNum;
- PathsImage(Map<String, Set<String>> pathImage, long curSeqNum) {
+ public PathsImage(Map<String, Set<String>> pathImage, long curSeqNum, long curImgNum) {
this.pathImage = pathImage;
this.curSeqNum = curSeqNum;
+ this.curImgNum = curImgNum;
}
public long getCurSeqNum() {
return curSeqNum;
}
+ public long getCurImgNum() {
+ return curImgNum;
+ }
+
public Map<String, Set<String>> getPathImage() {
return pathImage;
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/PermissionsImage.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/PermissionsImage.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/PermissionsImage.java
index f03e93f..6c74e19 100644
--- a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/PermissionsImage.java
+++ b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/PermissionsImage.java
@@ -36,8 +36,8 @@ public class PermissionsImage {
private final Map<String, Map<String, String>> privilegeImage;
private final long curSeqNum;
- PermissionsImage(Map<String, List<String>> roleImage,
- Map<String, Map<String, String>> privilegeImage, long curSeqNum) {
+ public PermissionsImage(Map<String, List<String>> roleImage,
+ Map<String, Map<String, String>> privilegeImage, long curSeqNum) {
this.roleImage = roleImage;
this.privilegeImage = privilegeImage;
this.curSeqNum = curSeqNum;
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/SentryStore.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/SentryStore.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/SentryStore.java
index 1402ab1..c6f3cc8 100644
--- a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/SentryStore.java
+++ b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/SentryStore.java
@@ -2539,9 +2539,10 @@ public class SentryStore {
// from HMS. It does not have corresponding delta update.
pm.setDetachAllOnCommit(false); // No need to detach objects
long curChangeID = getLastProcessedChangeIDCore(pm, MSentryPathChange.class);
- Map<String, Set<String>> pathImage = retrieveFullPathsImageCore(pm);
+ long curImageID = getCurrentAuthzPathsSnapshotID(pm);
+ Map<String, Set<String>> pathImage = retrieveFullPathsImageCore(pm, curImageID);
- return new PathsImage(pathImage, curChangeID);
+ return new PathsImage(pathImage, curChangeID, curImageID);
}
});
}
@@ -2552,8 +2553,7 @@ public class SentryStore {
*
* @return a mapping of hiveObj to < Paths >.
*/
- private Map<String, Set<String>> retrieveFullPathsImageCore(PersistenceManager pm) {
- long currentSnapshotID = getCurrentAuthzPathsSnapshotID(pm);
+ private Map<String, Set<String>> retrieveFullPathsImageCore(PersistenceManager pm, long currentSnapshotID) {
if (currentSnapshotID <= EMPTY_PATHS_SNAPSHOT_ID) {
return Collections.emptyMap();
}
@@ -3641,6 +3641,21 @@ public class SentryStore {
}
/**
+ * Gets the last processed HMS snapshot ID for path delta changes.
+ *
+ * @return latest path change ID.
+ */
+ public long getLastProcessedImageID() throws Exception {
+ return tm.executeTransaction(new TransactionBlock<Long>() {
+ @Override
+ public Long execute(PersistenceManager pm) throws Exception {
+ pm.setDetachAllOnCommit(false); // No need to detach objects
+ return getCurrentAuthzPathsSnapshotID(pm);
+ }
+ });
+ }
+
+ /**
* Get the MSentryPermChange object by ChangeID.
*
* @param changeID the given changeID.
http://git-wip-us.apache.org/repos/asf/sentry/blob/b9eca214/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/persistent/TestSentryStore.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/persistent/TestSentryStore.java b/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/persistent/TestSentryStore.java
index ac266fe..395cba6 100644
--- a/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/persistent/TestSentryStore.java
+++ b/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/persistent/TestSentryStore.java
@@ -2443,6 +2443,7 @@ public class TestSentryStore extends org.junit.Assert {
"/user/hive/warehouse/db2.db/table2.3"));
sentryStore.persistFullPathsImage(authzPaths);
PathsImage pathsImage = sentryStore.retrieveFullPathsImage();
+ assertEquals(1, pathsImage.getCurImgNum());
Map<String, Set<String>> pathImage = pathsImage.getPathImage();
assertEquals(3, pathImage.size());
for (Map.Entry<String, Set<String>> entry : pathImage.entrySet()) {
@@ -2610,6 +2611,7 @@ public class TestSentryStore extends org.junit.Assert {
@Test
public void testPersistAndReplaceANewPathsImage() throws Exception {
Map<String, Set<String>> authzPaths = new HashMap<>();
+ PathsImage pathsImage;
// First image to persist (this will be replaced later)
authzPaths.put("db1.table1", Sets.newHashSet("/user/hive/warehouse/db2.db/table1.1",
@@ -2617,6 +2619,8 @@ public class TestSentryStore extends org.junit.Assert {
authzPaths.put("db1.table2", Sets.newHashSet("/user/hive/warehouse/db2.db/table2.1",
"/user/hive/warehouse/db2.db/table2.2"));
sentryStore.persistFullPathsImage(authzPaths);
+ pathsImage = sentryStore.retrieveFullPathsImage();
+ assertEquals(1, pathsImage.getCurImgNum());
// Second image to persist (it should replace first image)
authzPaths.clear();
@@ -2628,7 +2632,8 @@ public class TestSentryStore extends org.junit.Assert {
"/another-warehouse/db2.db/table2.3"));
sentryStore.persistFullPathsImage(authzPaths);
- PathsImage pathsImage = sentryStore.retrieveFullPathsImage();
+ pathsImage = sentryStore.retrieveFullPathsImage();
+ assertEquals(2, pathsImage.getCurImgNum());
Map<String, Set<String>> pathImage = pathsImage.getPathImage();
assertEquals(3, pathImage.size());