SentrySyncHMSNotificationsPostEventListener extends MetaStoreEventListener { private void notifyHmsEvent(SentryHmsEvent event) { if(event.isMetastoreTransactionActive()) { return; } if (!shouldSyncEvent(event)) { event.setEventId(0L); } try (SentryPolicyServiceClient sentryClient = this.getSentryServiceClient()) { LOGGER.debug("Notifying sentry about Notification for {} (id: {})", event.getEventType(), event.getEventId()); long sentryLatestProcessedId = sentryClient.notifyHmsEvent( getUserName(), event.getEventId(), event.getEventType().toString(), event.getOwnerType(), event.getOwnerName(), event.getAuthorizable()); LOGGER.debug("Finished Notifying sentry about Notification for {} (id: {})", event.getEventType(), event.getEventId()); LOGGER.debug("Latest processed event ID returned by the Sentry server: {}", sentryLatestProcessedId); updateProcessedId(sentryLatestProcessedId); } catch (Exception e) { LOGGER.error("Encountered failure while notifying notification for {} (id: {})", event.getEventType(), event.getEventId(), e); } } SentrySyncHMSNotificationsPostEventListener(Configuration config); @Override void onCreateTable(CreateTableEvent tableEvent); @Override void onDropTable(DropTableEvent tableEvent); @Override void onAlterTable(AlterTableEvent tableEvent); @Override void onAddPartition(AddPartitionEvent partitionEvent); @Override void onDropPartition(DropPartitionEvent partitionEvent); @Override void onAlterPartition(AlterPartitionEvent partitionEvent); @Override void onCreateDatabase(CreateDatabaseEvent dbEvent); @Override void onDropDatabase(DropDatabaseEvent dbEvent);  }
@Test public void testSuccessfulEventsWithAnEventIdSyncNotifications() throws Exception { long eventId = 1; callAllEventsThatSynchronize(EventType.CREATE_DATABASE, SUCCESSFUL_STATUS, eventId++); Mockito.verify( mockSentryClient, Mockito.times(1) ).notifyHmsEvent(anyString(), eq(eventId-1), eq(EventType.CREATE_DATABASE.toString()), anyObject(), anyString(), eq(new TSentryAuthorizable(SERVER1))); callAllEventsThatSynchronize(EventType.DROP_DATABASE, SUCCESSFUL_STATUS, eventId++); Mockito.verify( mockSentryClient, Mockito.times(1) ).notifyHmsEvent(anyString(), eq(eventId-1), eq(EventType.DROP_DATABASE.toString()), anyObject(), anyString(), eq(new TSentryAuthorizable(SERVER1))); callAllEventsThatSynchronize(EventType.CREATE_TABLE, SUCCESSFUL_STATUS, eventId++); Mockito.verify( mockSentryClient, Mockito.times(1) ).notifyHmsEvent(anyString(), eq(eventId-1), eq(EventType.CREATE_TABLE.toString()), eq(TSentryPrincipalType.USER), anyString(), eq(new TSentryAuthorizable(SERVER1))); long latestEventId = callAllEventsThatSynchronize(EventType.DROP_TABLE, SUCCESSFUL_STATUS, eventId++); Mockito.verify( mockSentryClient, Mockito.times(1) ).notifyHmsEvent(anyString(), eq(eventId-1), eq(EventType.DROP_TABLE.toString()), eq(TSentryPrincipalType.USER), anyString(), eq(new TSentryAuthorizable(SERVER1))); Mockito.verify( mockSentryClient, Mockito.times((int) latestEventId) ).close(); Mockito.verifyNoMoreInteractions(mockSentryClient); }
@Test public void testSyncNotificationsWithNewLatestProcessedIdMayAvoidSyncingCalls() throws Exception { Mockito.doAnswer(new Answer<Long>() { @Override public Long answer(InvocationOnMock invocation) throws Throwable { TSentryHmsEventNotification notification = (TSentryHmsEventNotification) invocation.getArguments()[0]; return notification.getId() + 1; } }).when(mockSentryClient).notifyHmsEvent(anyString(), anyLong(), anyString(), anyObject(), anyString(), anyObject()); long latestEventId = callAllEventsThatSynchronize(SUCCESSFUL_STATUS, EVENT_ID_SET); verifyInvocations(); Mockito.verify( mockSentryClient, Mockito.times((int) latestEventId) ).close(); }