HMSFollower implements Runnable, AutoCloseable, PubSub.Subscriber { public void processNotifications(Collection<NotificationEvent> events) throws Exception { boolean isNotificationProcessed; if (events.isEmpty()) { return; } for (NotificationEvent event : events) { isNotificationProcessed = false; try { if (!isLeader()) { LOGGER.debug("Not processing notifications since not a leader"); return; } isNotificationProcessed = notificationProcessor.processNotificationEvent(event); } catch (Exception e) { if (e.getCause() instanceof JDODataStoreException) { LOGGER.info("Received JDO Storage Exception, Could be because of processing " + "duplicate notification"); if (event.getEventId() <= sentryStore.getLastProcessedNotificationID()) { LOGGER.error("Received event with Id: {} which is smaller then the ID " + "persisted in store", event.getEventId()); break; } } else { LOGGER.error("Processing the notification with ID:{} failed with exception {}", event.getEventId(), e); } } if (!isNotificationProcessed) { try { LOGGER.debug("Explicitly Persisting Notification ID = {} ", event.getEventId()); sentryStore.persistLastProcessedNotificationID(event.getEventId()); } catch (Exception failure) { LOGGER.error("Received exception while persisting the notification ID = {}", event.getEventId()); throw failure; } } wakeUpWaitingClientsForSync(event.getEventId()); } } HMSFollower(Configuration conf, SentryStoreInterface store, LeaderStatusMonitor leaderMonitor,
              HiveConnectionFactory hiveConnectionFactory); @VisibleForTesting  HMSFollower(Configuration conf, SentryStoreInterface store, LeaderStatusMonitor leaderMonitor,
              HiveConnectionFactory hiveConnectionFactory, String authServerName); @VisibleForTesting static boolean isConnectedToHms(); @Override void close(); @Override void run(); void processNotifications(Collection<NotificationEvent> events); @Override void onMessage(PubSub.Topic topic, String message);  }
@Test public void testCreateDatabase() throws Exception { String dbName = "db1"; NotificationEvent notificationEvent = new NotificationEvent(1, 0, EventMessage.EventType.CREATE_DATABASE.toString(), messageFactory.buildCreateDatabaseMessage(new Database(dbName, null, "hdfs: .toString()); List<NotificationEvent> events = new ArrayList<>(); events.add(notificationEvent); HMSFollower hmsFollower = new HMSFollower(configuration, sentryStore, null, hiveConnectionFactory, hiveInstance); hmsFollower.processNotifications(events); TSentryAuthorizable authorizable = new TSentryAuthorizable(hiveInstance); authorizable.setServer(hiveInstance); authorizable.setDb("db1"); verify(sentryStore, times(1)) .dropPrivilege(authorizable, NotificationProcessor.getPermUpdatableOnDrop(authorizable)); }
@Test public void testDropDatabase() throws Exception { String dbName = "db1"; NotificationEvent notificationEvent = new NotificationEvent(1, 0, EventMessage.EventType.DROP_DATABASE.toString(), messageFactory.buildDropDatabaseMessage(new Database(dbName, null, "hdfs: .toString()); List<NotificationEvent> events = new ArrayList<>(); events.add(notificationEvent); HMSFollower hmsFollower = new HMSFollower(configuration, sentryStore, null, hiveConnectionFactory, hiveInstance); hmsFollower.processNotifications(events); TSentryAuthorizable authorizable = new TSentryAuthorizable(hiveInstance); authorizable.setServer(hiveInstance); authorizable.setDb("db1"); verify(sentryStore, times(1)) .dropPrivilege(authorizable, NotificationProcessor.getPermUpdatableOnDrop(authorizable)); }
@Test public void testCreateTable() throws Exception { String dbName = "db1"; String tableName = "table1"; StorageDescriptor sd = new StorageDescriptor(); sd.setLocation("hdfs: NotificationEvent notificationEvent = new NotificationEvent(1, 0, EventMessage.EventType.CREATE_TABLE.toString(), messageFactory.buildCreateTableMessage( new Table(tableName, dbName, null, 0, 0, 0, sd, null, null, null, null, null), Collections.emptyIterator()).toString()); List<NotificationEvent> events = new ArrayList<>(); events.add(notificationEvent); HMSFollower hmsFollower = new HMSFollower(configuration, sentryStore, null, hiveConnectionFactory, hiveInstance); hmsFollower.processNotifications(events); TSentryAuthorizable authorizable = new TSentryAuthorizable(hiveInstance); authorizable.setServer(hiveInstance); authorizable.setDb("db1"); authorizable.setTable(tableName); verify(sentryStore, times(1)) .dropPrivilege(authorizable, NotificationProcessor.getPermUpdatableOnDrop(authorizable)); }
@Test public void testDropTable() throws Exception { String dbName = "db1"; String tableName = "table1"; StorageDescriptor sd = new StorageDescriptor(); sd.setLocation("hdfs: NotificationEvent notificationEvent = new NotificationEvent(1, 0, EventMessage.EventType.DROP_TABLE.toString(), messageFactory.buildDropTableMessage( new Table(tableName, dbName, null, 0, 0, 0, sd, null, null, null, null, null)) .toString()); List<NotificationEvent> events = new ArrayList<>(); events.add(notificationEvent); HMSFollower hmsFollower = new HMSFollower(configuration, sentryStore, null, hiveConnectionFactory, hiveInstance); hmsFollower.processNotifications(events); TSentryAuthorizable authorizable = new TSentryAuthorizable(hiveInstance); authorizable.setServer(hiveInstance); authorizable.setDb("db1"); authorizable.setTable(tableName); verify(sentryStore, times(1)) .dropPrivilege(authorizable, NotificationProcessor.getPermUpdatableOnDrop(authorizable)); }
@Test public void testRenameTable() throws Exception { String dbName = "db1"; String tableName = "table1"; String newDbName = "db1"; String newTableName = "table2"; StorageDescriptor sd = new StorageDescriptor(); sd.setLocation("hdfs: NotificationEvent notificationEvent = new NotificationEvent(1, 0, EventMessage.EventType.ALTER_TABLE.toString(), messageFactory.buildAlterTableMessage( new Table(tableName, dbName, null, 0, 0, 0, sd, null, null, null, null, null), new Table(newTableName, newDbName, null, 0, 0, 0, sd, null, null, null, null, null)) .toString()); notificationEvent.setDbName(newDbName); notificationEvent.setTableName(newTableName); List<NotificationEvent> events = new ArrayList<>(); events.add(notificationEvent); HMSFollower hmsFollower = new HMSFollower(configuration, sentryStore, null, hiveConnectionFactory, hiveInstance); hmsFollower.processNotifications(events); TSentryAuthorizable authorizable = new TSentryAuthorizable(hiveInstance); authorizable.setServer(hiveInstance); authorizable.setDb(dbName); authorizable.setTable(tableName); TSentryAuthorizable newAuthorizable = new TSentryAuthorizable(hiveInstance); authorizable.setServer(hiveInstance); newAuthorizable.setDb(newDbName); newAuthorizable.setTable(newTableName); verify(sentryStore, times(1)).renamePrivilege(authorizable, newAuthorizable, NotificationProcessor.getPermUpdatableOnRename(authorizable, newAuthorizable)); }
@Ignore @Test public void testAlterPartitionWithInvalidEvent() throws Exception { String dbName = "db1"; String tableName1 = "table1"; String tableName2 = "table2"; long inputEventId = 1; List<NotificationEvent> events = new ArrayList<>(); NotificationEvent notificationEvent; List<FieldSchema> partCols; StorageDescriptor sd; Mockito.doNothing().when(sentryStore).persistLastProcessedNotificationID(Mockito.anyLong()); Mockito.doNothing().when(sentryStore).addAuthzPathsMapping(Mockito.anyString(), Mockito.anyCollection(), Mockito.any(UniquePathsUpdate.class)); Configuration configuration = new Configuration(); HMSFollower hmsFollower = new HMSFollower(configuration, sentryStore, null, hiveConnectionFactory, hiveInstance); sd = new StorageDescriptor(); sd.setLocation("hdfs: partCols = new ArrayList<>(); partCols.add(new FieldSchema("ds", "string", "")); Table table = new Table(tableName1, dbName, null, 0, 0, 0, sd, partCols, null, null, null, null); notificationEvent = new NotificationEvent(inputEventId, 0, EventMessage.EventType.CREATE_TABLE.toString(), messageFactory.buildCreateTableMessage(table, Collections.emptyIterator()).toString()); notificationEvent.setDbName(dbName); notificationEvent.setTableName(tableName1); events.add(notificationEvent); inputEventId += 1; hmsFollower.processNotifications(events); verify(sentryStore, times(1)).addAuthzPathsMapping(Mockito.anyString(), Mockito.anyCollection(), Mockito.any(UniquePathsUpdate.class)); verify(sentryStore, times(0)).persistLastProcessedNotificationID(Mockito.anyLong()); reset(sentryStore); events.clear(); List<Partition> partitions = new ArrayList<>(); StorageDescriptor invalidSd = new StorageDescriptor(); invalidSd.setLocation(null); Partition partition = new Partition(Collections.singletonList("today"), dbName, tableName1, 0, 0, sd, null); partitions.add(partition); notificationEvent = new NotificationEvent(inputEventId, 0, EventType.ADD_PARTITION.toString(), messageFactory.buildAddPartitionMessage(table, partitions.iterator(), Collections.emptyIterator()).toString()); notificationEvent.setDbName(dbName); notificationEvent.setTableName(tableName1); events.add(notificationEvent); inputEventId += 1; hmsFollower.processNotifications(events); verify(sentryStore, times(1)).addAuthzPathsMapping(Mockito.anyString(), Mockito.anyCollection(), Mockito.any(UniquePathsUpdate.class)); verify(sentryStore, times(0)).persistLastProcessedNotificationID(Mockito.anyLong()); reset(sentryStore); events.clear(); notificationEvent = new NotificationEvent(inputEventId, 0, EventType.ALTER_PARTITION.toString(), messageFactory.buildAlterPartitionMessage(table, partition, partition).toString()); notificationEvent.setDbName(dbName); notificationEvent.setTableName(tableName1); events.add(notificationEvent); inputEventId += 1; hmsFollower.processNotifications(events); verify(sentryStore, times(1)).persistLastProcessedNotificationID(inputEventId - 1); reset(sentryStore); events.clear(); sd = new StorageDescriptor(); sd.setLocation("hdfs: Partition updatedPartition = new Partition(partition); updatedPartition.setSd(sd); notificationEvent = new NotificationEvent(inputEventId, 0, EventType.ALTER_PARTITION.toString(), messageFactory.buildAlterPartitionMessage(table, partition, updatedPartition).toString()); notificationEvent.setDbName(dbName); notificationEvent.setTableName(tableName1); events.add(notificationEvent); inputEventId += 1; hmsFollower.processNotifications(events); verify(sentryStore, times(1)).updateAuthzPathsMapping(Mockito.anyString(), Mockito.anyString(), Mockito.anyString(), Mockito.any(UniquePathsUpdate.class)); verify(sentryStore, times(0)).persistLastProcessedNotificationID(inputEventId - 1); reset(sentryStore); events.clear(); sd = new StorageDescriptor(); sd.setLocation("hdfs: partCols = new ArrayList<>(); partCols.add(new FieldSchema("ds", "string", "")); Table table1 = new Table(tableName2, dbName, null, 0, 0, 0, sd, partCols, null, null, null, null); notificationEvent = new NotificationEvent(inputEventId, 0, EventMessage.EventType.CREATE_TABLE.toString(), messageFactory.buildCreateTableMessage(table1, Collections.emptyIterator()).toString()); notificationEvent.setDbName(dbName); notificationEvent.setTableName(tableName2); events.add(notificationEvent); hmsFollower.processNotifications(events); verify(sentryStore, times(1)).addAuthzPathsMapping(Mockito.anyString(), Mockito.anyCollection(), Mockito.any(UniquePathsUpdate.class)); verify(sentryStore, times(0)).persistLastProcessedNotificationID(Mockito.anyLong()); }
@Test public void testAlterTableWithInvalidEvent() throws Exception { String dbName = "db1"; String tableName1 = "table1"; String tableName2 = "table2"; long inputEventId = 1; List<NotificationEvent> events = new ArrayList<>(); NotificationEvent notificationEvent; List<FieldSchema> partCols; StorageDescriptor sd; Mockito.doNothing().when(sentryStore).persistLastProcessedNotificationID(Mockito.anyLong()); Mockito.doNothing().when(sentryStore).addAuthzPathsMapping(Mockito.anyString(), Mockito.anyCollection(), Mockito.any(UniquePathsUpdate.class)); Configuration configuration = new Configuration(); enableHdfsSyncInSentry(configuration); HMSFollower hmsFollower = new HMSFollower(configuration, sentryStore, null, hiveConnectionFactory, hiveInstance); sd = new StorageDescriptor(); sd.setLocation("hdfs: partCols = new ArrayList<>(); partCols.add(new FieldSchema("ds", "string", "")); Table table = new Table(tableName1, dbName, null, 0, 0, 0, sd, partCols, null, null, null, null); notificationEvent = new NotificationEvent(inputEventId, 0, EventMessage.EventType.CREATE_TABLE.toString(), messageFactory.buildCreateTableMessage(table, Collections.emptyIterator()).toString()); notificationEvent.setDbName(dbName); notificationEvent.setTableName(tableName1); events.add(notificationEvent); inputEventId += 1; hmsFollower.processNotifications(events); verify(sentryStore, times(1)).addAuthzPathsMapping(Mockito.anyString(), Mockito.anyCollection(), Mockito.any(UniquePathsUpdate.class)); verify(sentryStore, times(0)).persistLastProcessedNotificationID(Mockito.anyLong()); reset(sentryStore); events.clear(); notificationEvent = new NotificationEvent(1, 0, EventMessage.EventType.ALTER_TABLE.toString(), messageFactory.buildAlterTableMessage( new Table(tableName1, dbName, null, 0, 0, 0, sd, null, null, null, null, null), new Table(tableName1, dbName, null, 0, 0, 0, sd, null, null, null, null, null)) .toString()); notificationEvent.setDbName(dbName); notificationEvent.setTableName(tableName1); events = new ArrayList<>(); events.add(notificationEvent); inputEventId += 1; hmsFollower.processNotifications(events); verify(sentryStore, times(0)).renameAuthzObj(Mockito.anyString(), Mockito.anyString(), Mockito.any(UniquePathsUpdate.class)); verify(sentryStore, times(0)).deleteAuthzPathsMapping(Mockito.anyString(), Mockito.anyCollection(), Mockito.any(UniquePathsUpdate.class)); verify(sentryStore, times(1)).persistLastProcessedNotificationID(Mockito.anyLong()); reset(sentryStore); events.clear(); sd = new StorageDescriptor(); sd.setLocation("hdfs: partCols = new ArrayList<>(); partCols.add(new FieldSchema("ds", "string", "")); Table table1 = new Table(tableName2, dbName, null, 0, 0, 0, sd, partCols, null, null, null, null); notificationEvent = new NotificationEvent(inputEventId, 0, EventMessage.EventType.CREATE_TABLE.toString(), messageFactory.buildCreateTableMessage(table1, Collections.emptyIterator()).toString()); notificationEvent.setDbName(dbName); notificationEvent.setTableName(tableName2); events.add(notificationEvent); hmsFollower.processNotifications(events); verify(sentryStore, times(1)).addAuthzPathsMapping(Mockito.anyString(), Mockito.anyCollection(), Mockito.any(UniquePathsUpdate.class)); verify(sentryStore, times(0)).persistLastProcessedNotificationID(Mockito.anyLong()); }
@Test public void testCreateTableAfterInvalidEvent() throws Exception { String dbName = "db1"; String tableName = "table1"; long inputEventId = 1; Mockito.doNothing().when(sentryStore).persistLastProcessedNotificationID(Mockito.anyLong()); Mockito.doNothing().when(sentryStore) .addAuthzPathsMapping(Mockito.anyString(), Mockito.anyCollection(), Mockito.any(UniquePathsUpdate.class)); StorageDescriptor invalidSd = new StorageDescriptor(); invalidSd.setLocation(null); NotificationEvent invalidNotificationEvent = new NotificationEvent(inputEventId, 0, EventMessage.EventType.CREATE_TABLE.toString(), messageFactory.buildCreateTableMessage( new Table(tableName, dbName, null, 0, 0, 0, invalidSd, null, null, null, null, null), Collections.emptyIterator()).toString()); StorageDescriptor sd = new StorageDescriptor(); sd.setLocation("hdfs: inputEventId += 1; NotificationEvent notificationEvent = new NotificationEvent(inputEventId, 0, EventMessage.EventType.CREATE_TABLE.toString(), messageFactory.buildCreateTableMessage( new Table(tableName, dbName, null, 0, 0, 0, sd, null, null, null, null, null), Collections.emptyIterator()).toString()); List<NotificationEvent> events = new ArrayList<>(); events.add(invalidNotificationEvent); events.add(notificationEvent); Configuration configuration = new Configuration(); enableHdfsSyncInSentry(configuration); HMSFollower hmsFollower = new HMSFollower(configuration, sentryStore, null, hiveConnectionFactory, hiveInstance); hmsFollower.processNotifications(events); verify(sentryStore, times(1)).persistLastProcessedNotificationID(inputEventId - 1); verify(sentryStore, times(1)).addAuthzPathsMapping(Mockito.anyString(), Mockito.anyCollection(), Mockito.any(UniquePathsUpdate.class)); }
@Test public void testNoHdfsSyncAlterTableIsPersisted() throws Exception { String dbName = "db1"; String tableName = "table1"; String newDbName = "db1"; String newTableName = "table2"; StorageDescriptor sd = new StorageDescriptor(); sd.setLocation("hdfs: NotificationEvent notificationEvent = new NotificationEvent(1, 0, EventMessage.EventType.ALTER_TABLE.toString(), messageFactory.buildAlterTableMessage( new Table(tableName, dbName, null, 0, 0, 0, sd, null, null, null, null, null), new Table(newTableName, newDbName, null, 0, 0, 0, sd, null, null, null, null, null)) .toString()); notificationEvent.setDbName(newDbName); notificationEvent.setTableName(newTableName); List<NotificationEvent> events = new ArrayList<>(); events.add(notificationEvent); Configuration configuration = new Configuration(); HMSFollower hmsFollower = new HMSFollower(configuration, sentryStore, null, hiveConnectionFactory, hiveInstance); hmsFollower.processNotifications(events); TSentryAuthorizable authorizable = new TSentryAuthorizable(hiveInstance); authorizable.setServer(hiveInstance); authorizable.setDb(dbName); authorizable.setTable(tableName); TSentryAuthorizable newAuthorizable = new TSentryAuthorizable(hiveInstance); authorizable.setServer(hiveInstance); newAuthorizable.setDb(newDbName); newAuthorizable.setTable(newTableName); verify(sentryStore, times(1)).renamePrivilege(authorizable, newAuthorizable, NotificationProcessor.getPermUpdatableOnRename(authorizable, newAuthorizable)); }