@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)); }
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 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 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); }
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); }
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);  }