This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch atlas-4464
in repository https://gitbox.apache.org/repos/asf/atlas.git
The following commit(s) were added to refs/heads/atlas-4464 by this push:
new fd1743d57 ATLAS-4464: Concurrent ingest implementation - fixing test
cases
fd1743d57 is described below
commit fd1743d57261ca5a819cb96db188e11919b844e2
Author: Radhika Kundam <[email protected]>
AuthorDate: Thu Jun 4 16:35:28 2026 -0700
ATLAS-4464: Concurrent ingest implementation - fixing test cases
---
.../notification/NotificationHookConsumerTest.java | 338 ++++++++++++---------
1 file changed, 198 insertions(+), 140 deletions(-)
diff --git
a/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerTest.java
b/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerTest.java
index b3653aa9f..4bbfe4109 100644
---
a/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerTest.java
+++
b/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerTest.java
@@ -51,6 +51,7 @@ import org.apache.atlas.type.AtlasEntityType;
import org.apache.atlas.type.AtlasStructType;
import org.apache.atlas.type.AtlasType;
import org.apache.atlas.type.AtlasTypeRegistry;
+import org.apache.atlas.util.AdaptiveWaiter;
import org.apache.atlas.util.AtlasMetricsUtil;
import org.apache.atlas.v1.model.instance.Referenceable;
import org.apache.atlas.v1.model.notification.HookNotificationV1;
@@ -65,6 +66,7 @@ import org.mockito.MockedStatic;
import org.mockito.MockitoAnnotations;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
+import org.slf4j.LoggerFactory;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -202,17 +204,23 @@ public class NotificationHookConsumerTest {
}
@Test
- public void testCommitIsNotCalledEvenWhenMessageProcessingFails() throws
AtlasServiceException, AtlasException, AtlasBaseException {
+ public void testCommitIsNotCalledEvenWhenMessageProcessingFails() throws
Exception {
+ Configuration config = buildFailedMsgCacheConfig(10);
+ when(config.getInt(NotificationHookConsumer.CONSUMER_RETRIES_PROPERTY,
3)).thenReturn(1);
+
NotificationHookConsumer notificationHookConsumer = new
NotificationHookConsumer(notificationInterface, atlasEntityStore, serviceState,
instanceConverter, typeRegistry, metricsUtil, null, asyncImporter);
NotificationConsumer consumer = mock(NotificationConsumer.class);
- NotificationHookConsumer.HookConsumer hookConsumer =
notificationHookConsumer.new HookConsumer(consumer);
+ NotificationHookConsumer.HookConsumer hookConsumer =
(NotificationHookConsumer.HookConsumer) createHookConsumerWithEntityProcessor(
+ notificationHookConsumer, consumer,
createEntityProcessor(config));
+
EntityCreateRequest message = new EntityCreateRequest("user",
Collections.singletonList(mock(Referenceable.class)));
when(atlasEntityStore.createOrUpdate(any(EntityStream.class),
anyBoolean())).thenThrow(new RuntimeException("Simulating exception in
processing message"));
hookConsumer.handleMessage(new AtlasKafkaMessage(message, -1,
KafkaNotification.ATLAS_HOOK_TOPIC, -1));
- verifyZeroInteractions(consumer);
+ // After max retries, the offset is committed so the consumer can move
past the failed message
+ verify(consumer, times(1)).commit(any(TopicPartition.class),
anyLong());
}
@Test
@@ -765,11 +773,13 @@ public class NotificationHookConsumerTest {
@Test
public void testHookConsumerCreateOrUpdateWithBatching() throws Exception {
- NotificationHookConsumer consumer = createTestConsumerWithBatching(5);
// batch size of 5
+ Configuration batchConfig = buildBatchingConfig(5);
+ NotificationHookConsumer consumer = createTestConsumer();
NotificationConsumer<HookNotification> notificationConsumer =
mock(NotificationConsumer.class);
- Object hookConsumer = createHookConsumer(consumer,
notificationConsumer);
+ Object hookConsumer = createHookConsumerWithEntityProcessor(consumer,
notificationConsumer, createEntityProcessor(batchConfig));
- Method createOrUpdateMethod =
hookConsumer.getClass().getDeclaredMethod("createOrUpdate",
AtlasEntitiesWithExtInfo.class, boolean.class,
AtlasMetricsUtil.NotificationStat.class, PreprocessorContext.class);
+ Object entityProcessor = getEntityProcessor(hookConsumer);
+ Method createOrUpdateMethod =
SerialEntityProcessor.class.getDeclaredMethod("createOrUpdate",
AtlasEntitiesWithExtInfo.class, boolean.class,
AtlasMetricsUtil.NotificationStat.class, PreprocessorContext.class);
createOrUpdateMethod.setAccessible(true);
// Create entities with more than batch size
@@ -796,7 +806,7 @@ public class NotificationHookConsumerTest {
when(mockResponse.getDeletedEntities()).thenReturn(Collections.emptyList());
when(atlasEntityStore.createOrUpdate(any(AtlasEntityStream.class),
eq(false))).thenReturn(mockResponse);
- createOrUpdateMethod.invoke(hookConsumer, entitiesWithExtInfo, false,
stats, context);
+ createOrUpdateMethod.invoke(entityProcessor, entitiesWithExtInfo,
false, stats, context);
verify(atlasEntityStore,
times(3)).createOrUpdate(any(AtlasEntityStream.class), eq(false));
}
@@ -807,7 +817,8 @@ public class NotificationHookConsumerTest {
NotificationConsumer<HookNotification> notificationConsumer =
mock(NotificationConsumer.class);
Object hookConsumer = createHookConsumer(consumer,
notificationConsumer);
- Method createOrUpdateMethod =
hookConsumer.getClass().getDeclaredMethod("createOrUpdate",
AtlasEntitiesWithExtInfo.class, boolean.class,
AtlasMetricsUtil.NotificationStat.class, PreprocessorContext.class);
+ Object entityProcessor = getEntityProcessor(hookConsumer);
+ Method createOrUpdateMethod =
SerialEntityProcessor.class.getDeclaredMethod("createOrUpdate",
AtlasEntitiesWithExtInfo.class, boolean.class,
AtlasMetricsUtil.NotificationStat.class, PreprocessorContext.class);
createOrUpdateMethod.setAccessible(true);
AtlasEntity entity = new AtlasEntity("TestType");
@@ -828,7 +839,7 @@ public class NotificationHookConsumerTest {
when(mockResponse.getDeletedEntities()).thenReturn(Collections.emptyList());
when(atlasEntityStore.createOrUpdate(any(AtlasEntityStream.class),
anyBoolean())).thenReturn(mockResponse);
- createOrUpdateMethod.invoke(hookConsumer, entitiesWithExtInfo, false,
stats, context);
+ createOrUpdateMethod.invoke(entityProcessor, entitiesWithExtInfo,
false, stats, context);
verify(atlasEntityStore,
times(2)).createOrUpdate(any(AtlasEntityStream.class), anyBoolean());
verify(context).prepareForPostUpdate();
@@ -836,59 +847,25 @@ public class NotificationHookConsumerTest {
@Test
public void testAdaptiveWaiterWithDifferentExceptions() throws Exception {
- NotificationHookConsumer consumer = createTestConsumer();
-
- Class<?> adaptiveWaiterClass =
getInnerClass(NotificationHookConsumer.class, "AdaptiveWaiter");
- Object adaptiveWaiter =
adaptiveWaiterClass.getDeclaredConstructor(long.class, long.class, long.class)
- .newInstance(100L, 5000L, 100L);
-
- Method pauseMethod = adaptiveWaiterClass.getDeclaredMethod("pause",
Throwable.class);
- pauseMethod.setAccessible(true);
+ AdaptiveWaiter adaptiveWaiter = new AdaptiveWaiter(100L, 5000L, 100L);
- // Test with different exception types
- pauseMethod.invoke(adaptiveWaiter, new RuntimeException("Test runtime
exception"));
- pauseMethod.invoke(adaptiveWaiter, new IllegalStateException("Test
illegal state"));
- pauseMethod.invoke(adaptiveWaiter, new AtlasBaseException("Test atlas
exception"));
+ adaptiveWaiter.pause(new RuntimeException("Test runtime exception"));
+ adaptiveWaiter.pause(new IllegalStateException("Test illegal state"));
+ adaptiveWaiter.pause(new AtlasBaseException("Test atlas exception"));
- Field waitDurationField =
adaptiveWaiterClass.getDeclaredField("waitDuration");
- waitDurationField.setAccessible(true);
- long waitDuration = (Long) waitDurationField.get(adaptiveWaiter);
-
- // Should have increased wait duration after multiple pauses
- assertTrue(waitDuration > 100L);
+ assertTrue(adaptiveWaiter.waitDuration > 100L);
}
@Test
public void testAdaptiveWaiterResetAfterLongInterval() throws Exception {
- NotificationHookConsumer consumer = createTestConsumer();
+ AdaptiveWaiter adaptiveWaiter = new AdaptiveWaiter(100L, 1000L, 100L);
- Class<?> adaptiveWaiterClass =
getInnerClass(NotificationHookConsumer.class, "AdaptiveWaiter");
- Object adaptiveWaiter =
adaptiveWaiterClass.getDeclaredConstructor(long.class, long.class, long.class)
- .newInstance(100L, 1000L, 100L);
+ adaptiveWaiter.pause(new RuntimeException("Test"));
+ Thread.sleep(2500); // resetInterval = maxDuration * 2
- Method pauseMethod = adaptiveWaiterClass.getDeclaredMethod("pause",
Throwable.class);
- pauseMethod.setAccessible(true);
+ adaptiveWaiter.pause(new RuntimeException("Test again"));
- Method setWaitDurationsMethod =
adaptiveWaiterClass.getDeclaredMethod("setWaitDurations");
- setWaitDurationsMethod.setAccessible(true);
-
- Field lastWaitAtField =
adaptiveWaiterClass.getDeclaredField("lastWaitAt");
- lastWaitAtField.setAccessible(true);
-
- // Simulate wait
- pauseMethod.invoke(adaptiveWaiter, new RuntimeException("Test"));
-
- // Set lastWaitAt to simulate long interval
- lastWaitAtField.set(adaptiveWaiter, System.currentTimeMillis() -
10000);
-
- // Should reset wait duration due to long interval
- setWaitDurationsMethod.invoke(adaptiveWaiter);
-
- Field waitDurationField =
adaptiveWaiterClass.getDeclaredField("waitDuration");
- waitDurationField.setAccessible(true);
- long waitDuration = (Long) waitDurationField.get(adaptiveWaiter);
-
- assertEquals(100L, waitDuration); // Should be reset to minimum
+ assertEquals(100L, adaptiveWaiter.waitDuration);
}
@Test
@@ -1090,7 +1067,7 @@ public class NotificationHookConsumerTest {
@Test
public void testComplexPreprocessingScenarios() throws Exception {
- NotificationHookConsumer consumer =
createTestConsumerWithPreprocessing();
+ Object entityProcessor = createEntityProcessorWithPreprocessing();
// Test complex entity with multiple relationships and attributes
AtlasEntity complexEntity = new AtlasEntity("hive_table");
@@ -1110,10 +1087,10 @@ public class NotificationHookConsumerTest {
EntityCreateRequestV2 createRequest = new
EntityCreateRequestV2("testUser", entities);
AtlasKafkaMessage<HookNotification> kafkaMsg = new
AtlasKafkaMessage<>(createRequest, 1L, "test-topic", 0);
- Method preProcessMethod =
NotificationHookConsumer.class.getDeclaredMethod("preProcessNotificationMessage",
AtlasKafkaMessage.class);
+ Method preProcessMethod =
SerialEntityProcessor.class.getDeclaredMethod("preProcessNotificationMessage",
AtlasKafkaMessage.class);
preProcessMethod.setAccessible(true);
- PreprocessorContext result = (PreprocessorContext)
preProcessMethod.invoke(consumer, kafkaMsg);
+ PreprocessorContext result = (PreprocessorContext)
preProcessMethod.invoke(entityProcessor, kafkaMsg);
assertNotNull(result);
verify(atlasEntityStore, never()).createOrUpdate(any(), anyBoolean());
// Just preprocessing
@@ -1143,9 +1120,10 @@ public class NotificationHookConsumerTest {
@Test
public void testHookConsumerMaxRetriesWithFailedMessageRecording() throws
Exception {
- NotificationHookConsumer consumer =
createTestConsumerWithFailedMsgCache(2); // small cache
+ Configuration cacheConfig = buildFailedMsgCacheConfig(2);
+ NotificationHookConsumer consumer = createTestConsumer();
NotificationConsumer<HookNotification> notificationConsumer =
mock(NotificationConsumer.class);
- Object hookConsumer = createHookConsumer(consumer,
notificationConsumer);
+ Object hookConsumer = createHookConsumerWithEntityProcessor(consumer,
notificationConsumer, createEntityProcessor(cacheConfig));
Method handleMessageMethod =
hookConsumer.getClass().getDeclaredMethod("handleMessage",
AtlasKafkaMessage.class);
handleMessageMethod.setAccessible(true);
@@ -1276,7 +1254,9 @@ public class NotificationHookConsumerTest {
handleMessageMethod.invoke(hookConsumer, kafkaMsg);
handleMessageMethod.invoke(hookConsumer, kafkaMsg);
- verify(notificationConsumer,
times(2)).commit(any(TopicPartition.class), anyLong());
+ // First message commits; second reuses the same offset and polls
instead of committing again
+ verify(notificationConsumer,
times(1)).commit(any(TopicPartition.class), anyLong());
+ verify(notificationConsumer, times(1)).poll();
}
@Test
@@ -1309,9 +1289,9 @@ public class NotificationHookConsumerTest {
@Test
public void testSkipHiveColumnLineageWithDuplicates() throws Exception {
- NotificationHookConsumer consumer =
createTestConsumerWithHiveLineageSkip();
+ Object entityProcessor = createEntityProcessorWithHiveLineageSkip();
- Method skipHiveLineageMethod =
NotificationHookConsumer.class.getDeclaredMethod("skipHiveColumnLineage",
PreprocessorContext.class);
+ Method skipHiveLineageMethod =
SerialEntityProcessor.class.getDeclaredMethod("skipHiveColumnLineage",
PreprocessorContext.class);
skipHiveLineageMethod.setAccessible(true);
PreprocessorContext context = mock(PreprocessorContext.class);
@@ -1332,7 +1312,7 @@ public class NotificationHookConsumerTest {
when(context.getKafkaMessageOffset()).thenReturn(100L);
when(context.getKafkaPartition()).thenReturn(1);
- skipHiveLineageMethod.invoke(consumer, context);
+ skipHiveLineageMethod.invoke(entityProcessor, context);
verify(context, atLeast(1)).getEntities();
}
@@ -1411,7 +1391,7 @@ public class NotificationHookConsumerTest {
public void testEntityUpdateWithComplexRelationships() throws Exception {
NotificationHookConsumer consumer = createTestConsumer();
- Method updateReferencesMethod =
NotificationHookConsumer.class.getDeclaredMethod("updateProcessedEntityReferences",
List.class, Map.class);
+ Method updateReferencesMethod =
SerialEntityProcessor.class.getDeclaredMethod("updateProcessedEntityReferences",
List.class, Map.class);
updateReferencesMethod.setAccessible(true);
// Create entities with complex relationships
@@ -1448,7 +1428,7 @@ public class NotificationHookConsumerTest {
guidAssignments.put("old-col1-guid", "new-col1-guid");
guidAssignments.put("old-schema-guid", "new-schema-guid");
- updateReferencesMethod.invoke(consumer, entities, guidAssignments);
+ updateReferencesMethod.invoke(getEntityProcessor(consumer), entities,
guidAssignments);
// Verify references were updated
assertEquals("new-db-guid", ((AtlasObjectId)
entity1.getAttribute("database")).getGuid());
@@ -1545,7 +1525,7 @@ public class NotificationHookConsumerTest {
@Test
public void testPreProcessNotificationMessage() throws Exception {
- NotificationHookConsumer consumer =
createTestConsumerWithPreprocessing();
+ Object entityProcessor = createEntityProcessorWithPreprocessing();
// Create test message
AtlasEntity entity = new AtlasEntity("hive_table");
@@ -1559,10 +1539,10 @@ public class NotificationHookConsumerTest {
new AtlasKafkaMessage<>(createRequest, 1L, "test-topic", 0);
Method preProcessMethod =
-
NotificationHookConsumer.class.getDeclaredMethod("preProcessNotificationMessage",
AtlasKafkaMessage.class);
+
SerialEntityProcessor.class.getDeclaredMethod("preProcessNotificationMessage",
AtlasKafkaMessage.class);
preProcessMethod.setAccessible(true);
- PreprocessorContext result = (PreprocessorContext)
preProcessMethod.invoke(consumer, kafkaMsg);
+ PreprocessorContext result = (PreprocessorContext)
preProcessMethod.invoke(entityProcessor, kafkaMsg);
assertNotNull(result, "Preprocessing should return a valid
PreprocessorContext");
}
@@ -1571,9 +1551,9 @@ public class NotificationHookConsumerTest {
@Test
public void testPreprocessEntities() throws Exception {
- NotificationHookConsumer consumer =
createTestConsumerWithPreprocessing();
+ Object entityProcessor = createEntityProcessorWithPreprocessing();
- Method preprocessEntitiesMethod =
NotificationHookConsumer.class.getDeclaredMethod("preprocessEntities",
PreprocessorContext.class);
+ Method preprocessEntitiesMethod =
SerialEntityProcessor.class.getDeclaredMethod("preprocessEntities",
PreprocessorContext.class);
preprocessEntitiesMethod.setAccessible(true);
// Create mock context
@@ -1587,7 +1567,7 @@ public class NotificationHookConsumerTest {
when(context.getReferredEntities()).thenReturn(new HashMap<>());
when(context.isIgnoredEntity("test-guid")).thenReturn(false);
- preprocessEntitiesMethod.invoke(consumer, context);
+ preprocessEntitiesMethod.invoke(entityProcessor, context);
verify(context).getEntities();
}
@@ -1596,13 +1576,13 @@ public class NotificationHookConsumerTest {
public void testTrimAndPurgeMethod() throws Exception {
NotificationHookConsumer consumer = createTestConsumer();
- Method trimAndPurgeMethod =
NotificationHookConsumer.class.getDeclaredMethod("trimAndPurge",
String[].class, String.class);
+ Method trimAndPurgeMethod =
SerialEntityProcessor.class.getDeclaredMethod("trimAndPurge", String[].class,
String.class);
trimAndPurgeMethod.setAccessible(true);
// Test with valid values
String[] input = {" value1 ", "", " value2", null, "value3 "};
@SuppressWarnings("unchecked")
- List<String> result = (List<String>)
trimAndPurgeMethod.invoke(consumer, input, "default");
+ List<String> result = (List<String>)
trimAndPurgeMethod.invoke(getEntityProcessor(consumer), input, "default");
assertEquals(3, result.size());
assertTrue(result.contains("value1"));
@@ -1611,13 +1591,13 @@ public class NotificationHookConsumerTest {
// Test with null input
@SuppressWarnings("unchecked")
- List<String> defaultResult = (List<String>)
trimAndPurgeMethod.invoke(consumer, null, "default");
+ List<String> defaultResult = (List<String>)
trimAndPurgeMethod.invoke(getEntityProcessor(consumer), null, "default");
assertEquals(1, defaultResult.size());
assertEquals("default", defaultResult.get(0));
// Test with empty default
@SuppressWarnings("unchecked")
- List<String> emptyResult = (List<String>)
trimAndPurgeMethod.invoke(consumer, null, null);
+ List<String> emptyResult = (List<String>)
trimAndPurgeMethod.invoke(getEntityProcessor(consumer), null, null);
assertTrue(emptyResult.isEmpty());
}
@@ -1625,7 +1605,7 @@ public class NotificationHookConsumerTest {
public void testGetAuthenticationForUser() throws Exception {
NotificationHookConsumer consumer = createTestConsumer();
- Method getAuthMethod =
NotificationHookConsumer.class.getDeclaredMethod("getAuthenticationForUser",
String.class);
+ Method getAuthMethod =
SerialEntityProcessor.class.getDeclaredMethod("getAuthenticationForUser",
String.class);
getAuthMethod.setAccessible(true);
try
(MockedStatic<org.apache.atlas.web.security.AtlasAbstractAuthenticationProvider>
authProvider =
@@ -1633,14 +1613,14 @@ public class NotificationHookConsumerTest {
authProvider.when(() ->
org.apache.atlas.web.security.AtlasAbstractAuthenticationProvider.getAuthoritiesFromUGI("testUser"))
.thenReturn(new ArrayList<>());
- Object auth = getAuthMethod.invoke(consumer, "testUser");
+ Object auth = getAuthMethod.invoke(getEntityProcessor(consumer),
"testUser");
assertNotNull(auth);
// Test with null/empty username
- Object nullAuth = getAuthMethod.invoke(consumer, (String) null);
+ Object nullAuth =
getAuthMethod.invoke(getEntityProcessor(consumer), (String) null);
assertNull(nullAuth);
- Object emptyAuth = getAuthMethod.invoke(consumer, "");
+ Object emptyAuth =
getAuthMethod.invoke(getEntityProcessor(consumer), "");
assertNull(emptyAuth);
}
}
@@ -1649,7 +1629,7 @@ public class NotificationHookConsumerTest {
public void testSetCurrentUser() throws Exception {
NotificationHookConsumer consumer = createTestConsumer();
- Method setCurrentUserMethod =
NotificationHookConsumer.class.getDeclaredMethod("setCurrentUser",
String.class);
+ Method setCurrentUserMethod =
SerialEntityProcessor.class.getDeclaredMethod("setCurrentUser", String.class);
setCurrentUserMethod.setAccessible(true);
try
(MockedStatic<org.apache.atlas.web.security.AtlasAbstractAuthenticationProvider>
authProvider =
@@ -1664,7 +1644,7 @@ public class NotificationHookConsumerTest {
securityContext.when(org.springframework.security.core.context.SecurityContextHolder::getContext)
.thenReturn(mockContext);
- setCurrentUserMethod.invoke(consumer, "testUser");
+ setCurrentUserMethod.invoke(getEntityProcessor(consumer),
"testUser");
verify(mockContext).setAuthentication(any());
}
@@ -1674,7 +1654,7 @@ public class NotificationHookConsumerTest {
public void testUpdateProcessedEntityReferencesAtlasObjectId() throws
Exception {
NotificationHookConsumer consumer = createTestConsumer();
- Method updateMethod =
NotificationHookConsumer.class.getDeclaredMethod("updateProcessedEntityReferences",
AtlasObjectId.class, Map.class);
+ Method updateMethod =
SerialEntityProcessor.class.getDeclaredMethod("updateProcessedEntityReferences",
AtlasObjectId.class, Map.class);
updateMethod.setAccessible(true);
AtlasObjectId objectId = new AtlasObjectId("original-guid",
"TestType");
@@ -1691,7 +1671,7 @@ public class NotificationHookConsumerTest {
public void testUpdateProcessedEntityReferencesMap() throws Exception {
NotificationHookConsumer consumer = createTestConsumer();
- Method updateMethod =
NotificationHookConsumer.class.getDeclaredMethod("updateProcessedEntityReferences",
Map.class, Map.class);
+ Method updateMethod =
SerialEntityProcessor.class.getDeclaredMethod("updateProcessedEntityReferences",
Map.class, Map.class);
updateMethod.setAccessible(true);
Map<String, Object> objIdMap = new HashMap<>();
@@ -1702,7 +1682,7 @@ public class NotificationHookConsumerTest {
Map<String, String> guidAssignments = new HashMap<>();
guidAssignments.put("original-guid", "new-guid");
- updateMethod.invoke(consumer, objIdMap, guidAssignments);
+ updateMethod.invoke(getEntityProcessor(consumer), objIdMap,
guidAssignments);
assertEquals("new-guid", objIdMap.get("guid"));
assertFalse(objIdMap.containsKey("typeName"));
@@ -1713,7 +1693,7 @@ public class NotificationHookConsumerTest {
public void testUpdateProcessedEntityReferencesCollection() throws
Exception {
NotificationHookConsumer consumer = createTestConsumer();
- Method updateMethod =
NotificationHookConsumer.class.getDeclaredMethod("updateProcessedEntityReferences",
Collection.class, Map.class);
+ Method updateMethod =
SerialEntityProcessor.class.getDeclaredMethod("updateProcessedEntityReferences",
Collection.class, Map.class);
updateMethod.setAccessible(true);
AtlasObjectId objectId1 = new AtlasObjectId("guid1", "TestType1");
@@ -1729,7 +1709,7 @@ public class NotificationHookConsumerTest {
assertEquals("guid2", objectId2.getGuid());
assertEquals("TestType2", objectId2.getTypeName());
- updateMethod.invoke(consumer, collection, guidAssignments);
+ updateMethod.invoke(getEntityProcessor(consumer), collection,
guidAssignments);
// objectId1 should be updated, objectId2 should remain unchanged
assertEquals("new-guid1", objectId1.getGuid());
@@ -1742,7 +1722,7 @@ public class NotificationHookConsumerTest {
public void testRecordProcessedEntities() throws Exception {
NotificationHookConsumer consumer = createTestConsumer();
- Method recordMethod =
NotificationHookConsumer.class.getDeclaredMethod("recordProcessedEntities",
EntityMutationResponse.class, AtlasMetricsUtil.NotificationStat.class,
PreprocessorContext.class);
+ Method recordMethod =
SerialEntityProcessor.class.getDeclaredMethod("recordProcessedEntities",
EntityMutationResponse.class, AtlasMetricsUtil.NotificationStat.class,
PreprocessorContext.class);
recordMethod.setAccessible(true);
EntityMutationResponse mutationResponse =
mock(EntityMutationResponse.class);
@@ -1756,7 +1736,7 @@ public class NotificationHookConsumerTest {
when(context.getCreatedEntities()).thenReturn(Collections.emptySet());
when(context.getDeletedEntities()).thenReturn(Collections.emptySet());
- recordMethod.invoke(consumer, mutationResponse, stats, context);
+ recordMethod.invoke(getEntityProcessor(consumer), mutationResponse,
stats, context);
verify(stats).updateStats(mutationResponse);
verify(context).getGuidAssignments();
@@ -1766,28 +1746,28 @@ public class NotificationHookConsumerTest {
public void testIsEmptyMessageForDifferentTypes() throws Exception {
NotificationHookConsumer consumer = createTestConsumer();
- Method isEmptyMethod =
NotificationHookConsumer.class.getDeclaredMethod("isEmptyMessage",
AtlasKafkaMessage.class);
+ Method isEmptyMethod =
SerialEntityProcessor.class.getDeclaredMethod("isEmptyMessage",
AtlasKafkaMessage.class);
isEmptyMethod.setAccessible(true);
// Test CREATE_V2 with empty entities
EntityCreateRequestV2 emptyCreate = new EntityCreateRequestV2("user",
new AtlasEntitiesWithExtInfo());
AtlasKafkaMessage<HookNotification> emptyCreateMsg = new
AtlasKafkaMessage<>(emptyCreate, 1L, "topic", 0);
- assertTrue((Boolean) isEmptyMethod.invoke(consumer, emptyCreateMsg));
+ assertTrue((Boolean)
isEmptyMethod.invoke(getEntityProcessor(consumer), emptyCreateMsg));
// Test CREATE_V2 with null entities
EntityCreateRequestV2 nullCreate = new EntityCreateRequestV2("user",
null);
AtlasKafkaMessage<HookNotification> nullCreateMsg = new
AtlasKafkaMessage<>(nullCreate, 1L, "topic", 0);
- assertTrue((Boolean) isEmptyMethod.invoke(consumer, nullCreateMsg));
+ assertTrue((Boolean)
isEmptyMethod.invoke(getEntityProcessor(consumer), nullCreateMsg));
// Test UPDATE_V2 with empty entities
EntityUpdateRequestV2 emptyUpdate = new EntityUpdateRequestV2("user",
new AtlasEntitiesWithExtInfo());
AtlasKafkaMessage<HookNotification> emptyUpdateMsg = new
AtlasKafkaMessage<>(emptyUpdate, 1L, "topic", 0);
- assertTrue((Boolean) isEmptyMethod.invoke(consumer, emptyUpdateMsg));
+ assertTrue((Boolean)
isEmptyMethod.invoke(getEntityProcessor(consumer), emptyUpdateMsg));
// Test other message types (should return false)
EntityDeleteRequestV2 deleteRequest = new
EntityDeleteRequestV2("user", Collections.emptyList());
AtlasKafkaMessage<HookNotification> deleteMsg = new
AtlasKafkaMessage<>(deleteRequest, 1L, "topic", 0);
- assertFalse((Boolean) isEmptyMethod.invoke(consumer, deleteMsg));
+ assertFalse((Boolean)
isEmptyMethod.invoke(getEntityProcessor(consumer), deleteMsg));
}
@Test
@@ -1863,18 +1843,19 @@ public class NotificationHookConsumerTest {
NotificationConsumer<HookNotification> notificationConsumer =
mock(NotificationConsumer.class);
Object hookConsumer = createHookConsumer(consumer,
notificationConsumer);
- Method recordFailedMethod =
hookConsumer.getClass().getDeclaredMethod("recordFailedMessages");
+ Object entityProcessor = getEntityProcessor(hookConsumer);
+ Method recordFailedMethod =
SerialEntityProcessor.class.getDeclaredMethod("recordFailedMessages",
List.class);
recordFailedMethod.setAccessible(true);
// Add some failed messages
- Field failedMessagesField =
hookConsumer.getClass().getDeclaredField("failedMessages");
+ Field failedMessagesField =
SerialEntityProcessor.class.getDeclaredField("failedMessages");
failedMessagesField.setAccessible(true);
@SuppressWarnings("unchecked")
- List<String> failedMessages = (List<String>)
failedMessagesField.get(hookConsumer);
+ List<String> failedMessages = (List<String>)
failedMessagesField.get(entityProcessor);
failedMessages.add("failed message 1");
failedMessages.add("failed message 2");
- recordFailedMethod.invoke(hookConsumer);
+ recordFailedMethod.invoke(entityProcessor, failedMessages);
assertTrue(failedMessages.isEmpty()); // Should be cleared after
recording
}
@@ -2021,27 +2002,15 @@ public class NotificationHookConsumerTest {
public void testConstructorWithExceptionInConfigurationAccess() throws
Exception {
Configuration faultyConfig = mock(Configuration.class);
- // Make config throw error on access
+
when(faultyConfig.getInt(NotificationHookConsumer.CONSUMER_FAILEDCACHESIZE_PROPERTY,
1)).thenReturn(1);
when(faultyConfig.getInt(NotificationHookConsumer.CONSUMER_RETRIES_PROPERTY, 3))
.thenThrow(new RuntimeException("Config access error"));
- // Add other config properties to prevent NPE
-
when(faultyConfig.getBoolean("atlas.notification.create.shell.entity.for.non.existing.ref",
false))
- .thenReturn(false);
-
when(faultyConfig.getInt("atlas.notification.hook.consumer.buffering.interval",
10))
- .thenReturn(10);
-
when(faultyConfig.getInt("atlas.notification.hook.consumer.buffering.batch.size",
25))
- .thenReturn(25);
-
- try (MockedStatic<ApplicationProperties> appProps =
mockStatic(ApplicationProperties.class)) {
- appProps.when(ApplicationProperties::get).thenReturn(faultyConfig);
- try {
- new NotificationHookConsumer(notificationInterface,
atlasEntityStore, serviceState,
- instanceConverter, typeRegistry, metricsUtil,
entityCorrelationStore, asyncImporter);
- fail("Expected exception due to config access error");
- } catch (Exception e) {
- assertTrue(e.getMessage().contains("Config access error") ||
e.getCause().getMessage().contains("Config access error"));
- }
+ try {
+ createEntityProcessor(faultyConfig);
+ fail("Expected exception due to config access error");
+ } catch (RuntimeException e) {
+ assertTrue(e.getMessage().contains("Config access error"));
}
}
@@ -2127,9 +2096,9 @@ public class NotificationHookConsumerTest {
@Test
public void testPreprocessHiveTypes() throws Exception {
- NotificationHookConsumer consumer =
createTestConsumerWithPreprocessing();
+ Object entityProcessor = createEntityProcessorWithPreprocessing();
- Method preprocessHiveTypesMethod =
NotificationHookConsumer.class.getDeclaredMethod("preprocessHiveTypes",
PreprocessorContext.class);
+ Method preprocessHiveTypesMethod =
SerialEntityProcessor.class.getDeclaredMethod("preprocessHiveTypes",
PreprocessorContext.class);
preprocessHiveTypesMethod.setAccessible(true);
PreprocessorContext context = mock(PreprocessorContext.class);
@@ -2140,7 +2109,7 @@ public class NotificationHookConsumerTest {
List<AtlasEntity> entities = Collections.singletonList(hiveTable);
when(context.getEntities()).thenReturn(entities);
- preprocessHiveTypesMethod.invoke(consumer, context);
+ preprocessHiveTypesMethod.invoke(entityProcessor, context);
verify(context).getEntities();
}
@@ -2149,9 +2118,9 @@ public class NotificationHookConsumerTest {
@Test
public void testSkipHiveColumnLineage() throws Exception {
- NotificationHookConsumer consumer =
createTestConsumerWithPreprocessing();
+ Object entityProcessor = createEntityProcessorWithPreprocessing();
- Method skipHiveLineageMethod =
NotificationHookConsumer.class.getDeclaredMethod("skipHiveColumnLineage",
PreprocessorContext.class);
+ Method skipHiveLineageMethod =
SerialEntityProcessor.class.getDeclaredMethod("skipHiveColumnLineage",
PreprocessorContext.class);
skipHiveLineageMethod.setAccessible(true);
PreprocessorContext context = mock(PreprocessorContext.class);
@@ -2169,16 +2138,16 @@ public class NotificationHookConsumerTest {
when(context.getReferredEntities()).thenReturn(new HashMap<>());
// This is a void method, so just verify it executes without exception
- skipHiveLineageMethod.invoke(consumer, context);
+ skipHiveLineageMethod.invoke(entityProcessor, context);
verify(context, atLeast(1)).getEntities();
}
@Test
public void testRdbmsTypeRemoveOwnedRefAttrs() throws Exception {
- NotificationHookConsumer consumer =
createTestConsumerWithPreprocessing();
+ Object entityProcessor = createEntityProcessorWithPreprocessing();
- Method rdbmsRemoveMethod =
NotificationHookConsumer.class.getDeclaredMethod("rdbmsTypeRemoveOwnedRefAttrs",
PreprocessorContext.class);
+ Method rdbmsRemoveMethod =
SerialEntityProcessor.class.getDeclaredMethod("rdbmsTypeRemoveOwnedRefAttrs",
PreprocessorContext.class);
rdbmsRemoveMethod.setAccessible(true);
PreprocessorContext context = mock(PreprocessorContext.class);
@@ -2189,16 +2158,16 @@ public class NotificationHookConsumerTest {
List<AtlasEntity> entities = Collections.singletonList(rdbmsTable);
when(context.getEntities()).thenReturn(entities);
- rdbmsRemoveMethod.invoke(consumer, context);
+ rdbmsRemoveMethod.invoke(entityProcessor, context);
verify(context).getEntities();
}
@Test
public void testPruneObjectPrefixForS3V2Directory() throws Exception {
- NotificationHookConsumer consumer =
createTestConsumerWithPreprocessing();
+ Object entityProcessor = createEntityProcessorWithPreprocessing();
- Method pruneMethod =
NotificationHookConsumer.class.getDeclaredMethod("pruneObjectPrefixForS3V2Directory",
PreprocessorContext.class);
+ Method pruneMethod =
SerialEntityProcessor.class.getDeclaredMethod("pruneObjectPrefixForS3V2Directory",
PreprocessorContext.class);
pruneMethod.setAccessible(true);
PreprocessorContext context = mock(PreprocessorContext.class);
@@ -2208,16 +2177,16 @@ public class NotificationHookConsumerTest {
List<AtlasEntity> entities = Collections.singletonList(s3Object);
when(context.getEntities()).thenReturn(entities);
- pruneMethod.invoke(consumer, context);
+ pruneMethod.invoke(entityProcessor, context);
verify(context, atLeast(1)).getEntities();
}
@Test
public void testPreprocessSparkProcessAttributes() throws Exception {
- NotificationHookConsumer consumer =
createTestConsumerWithPreprocessing();
+ Object entityProcessor = createEntityProcessorWithPreprocessing();
- Method sparkPreprocessMethod =
NotificationHookConsumer.class.getDeclaredMethod("preprocessSparkProcessAttributes",
PreprocessorContext.class);
+ Method sparkPreprocessMethod =
SerialEntityProcessor.class.getDeclaredMethod("preprocessSparkProcessAttributes",
PreprocessorContext.class);
sparkPreprocessMethod.setAccessible(true);
PreprocessorContext context = mock(PreprocessorContext.class);
@@ -2227,7 +2196,7 @@ public class NotificationHookConsumerTest {
List<AtlasEntity> entities = Collections.singletonList(sparkProcess);
when(context.getEntities()).thenReturn(entities);
- sparkPreprocessMethod.invoke(consumer, context);
+ sparkPreprocessMethod.invoke(entityProcessor, context);
verify(context).getEntities();
}
@@ -2236,7 +2205,7 @@ public class NotificationHookConsumerTest {
public void testUpdateProcessedEntityReferencesObject() throws Exception {
NotificationHookConsumer consumer = createTestConsumer();
- Method updateMethod =
NotificationHookConsumer.class.getDeclaredMethod("updateProcessedEntityReferences",
Object.class, Map.class);
+ Method updateMethod =
SerialEntityProcessor.class.getDeclaredMethod("updateProcessedEntityReferences",
Object.class, Map.class);
updateMethod.setAccessible(true);
// Test with AtlasObjectId
@@ -2248,7 +2217,7 @@ public class NotificationHookConsumerTest {
assertEquals("original-guid", objectId.getGuid());
assertEquals("TestType", objectId.getTypeName());
- updateMethod.invoke(consumer, objectId, guidAssignments);
+ updateMethod.invoke(getEntityProcessor(consumer), objectId,
guidAssignments);
assertEquals("new-guid", objectId.getGuid());
assertNull(objectId.getTypeName()); // cleared after GUID assignment
@@ -2258,14 +2227,14 @@ public class NotificationHookConsumerTest {
objIdMap.put("typeName", "MapType");
guidAssignments.put("map-guid", "new-map-guid");
- updateMethod.invoke(consumer, objIdMap, guidAssignments);
+ updateMethod.invoke(getEntityProcessor(consumer), objIdMap,
guidAssignments);
assertEquals("new-map-guid", objIdMap.get("guid"));
// Test with Collection
List<AtlasObjectId> collection = Collections.singletonList(new
AtlasObjectId("Type1", "collection-guid"));
guidAssignments.put("collection-guid", "new-collection-guid");
- updateMethod.invoke(consumer, collection, guidAssignments);
+ updateMethod.invoke(getEntityProcessor(consumer), collection,
guidAssignments);
}
@Test
@@ -2430,7 +2399,7 @@ public class NotificationHookConsumerTest {
public void testSetCurrentUserWithAuthorizationEnabled() throws Exception {
NotificationHookConsumer consumer =
createTestConsumerWithAuthorization();
- Method setCurrentUserMethod =
NotificationHookConsumer.class.getDeclaredMethod("setCurrentUser",
String.class);
+ Method setCurrentUserMethod =
SerialEntityProcessor.class.getDeclaredMethod("setCurrentUser", String.class);
setCurrentUserMethod.setAccessible(true);
try
(MockedStatic<org.apache.atlas.web.security.AtlasAbstractAuthenticationProvider>
authProvider =
@@ -2445,7 +2414,7 @@ public class NotificationHookConsumerTest {
securityContext.when(org.springframework.security.core.context.SecurityContextHolder::getContext)
.thenReturn(mockContext);
- setCurrentUserMethod.invoke(consumer, "testUser");
+ setCurrentUserMethod.invoke(getEntityProcessor(consumer),
"testUser");
verify(mockContext).setAuthentication(any());
}
@@ -2457,7 +2426,7 @@ public class NotificationHookConsumerTest {
public void testGetAuthenticationForUserWithCache() throws Exception {
NotificationHookConsumer consumer =
createTestConsumerWithAuthorization();
- Method getAuthMethod =
NotificationHookConsumer.class.getDeclaredMethod("getAuthenticationForUser",
String.class);
+ Method getAuthMethod =
SerialEntityProcessor.class.getDeclaredMethod("getAuthenticationForUser",
String.class);
getAuthMethod.setAccessible(true);
try
(MockedStatic<org.apache.atlas.web.security.AtlasAbstractAuthenticationProvider>
authProvider =
@@ -2466,11 +2435,11 @@ public class NotificationHookConsumerTest {
.thenReturn(new ArrayList<>());
// First call - should cache
- Object auth1 = getAuthMethod.invoke(consumer, "cachedUser");
+ Object auth1 = getAuthMethod.invoke(getEntityProcessor(consumer),
"cachedUser");
assertNotNull(auth1);
// Second call - should use cache
- Object auth2 = getAuthMethod.invoke(consumer, "cachedUser");
+ Object auth2 = getAuthMethod.invoke(getEntityProcessor(consumer),
"cachedUser");
assertNotNull(auth2);
// Should only call the static method once due to caching
@@ -2483,7 +2452,7 @@ public class NotificationHookConsumerTest {
public void testIsEmptyMessageWithNullMessage() throws Exception {
NotificationHookConsumer consumer = createTestConsumer();
- Method isEmptyMethod = NotificationHookConsumer.class
+ Method isEmptyMethod = SerialEntityProcessor.class
.getDeclaredMethod("isEmptyMessage", AtlasKafkaMessage.class);
isEmptyMethod.setAccessible(true);
@@ -2491,7 +2460,7 @@ public class NotificationHookConsumerTest {
new AtlasKafkaMessage<>(null, 1L, "topic", 0);
try {
- isEmptyMethod.invoke(consumer, nullMessage);
+ isEmptyMethod.invoke(getEntityProcessor(consumer), nullMessage);
fail("Expected NullPointerException but none thrown");
} catch (InvocationTargetException e) {
assertTrue(e.getCause() instanceof NullPointerException,
@@ -2587,6 +2556,95 @@ public class NotificationHookConsumerTest {
}
}
+ private Configuration buildBatchingConfig(int batchSize) {
+ Configuration batchConfig = mock(Configuration.class);
+
when(batchConfig.getInt(NotificationHookConsumer.CONSUMER_COMMIT_BATCH_SIZE,
50)).thenReturn(batchSize);
+
when(batchConfig.getInt(NotificationHookConsumer.CONSUMER_RETRIES_PROPERTY,
3)).thenReturn(3);
+
when(batchConfig.getBoolean("atlas.notification.create.shell.entity.for.non.existing.ref",
false)).thenReturn(false);
+
when(batchConfig.getInt("atlas.notification.hook.consumer.buffering.interval",
10)).thenReturn(10);
+
when(batchConfig.getInt("atlas.notification.hook.consumer.buffering.batch.size",
25)).thenReturn(25);
+
when(batchConfig.getBoolean("atlas.notification.consumer.create.shell.entity.for.non-existing.ref",
true)).thenReturn(false);
+
when(batchConfig.getInt("atlas.notification.consumer.message.buffering.interval.seconds",
15)).thenReturn(10);
+
when(batchConfig.getInt("atlas.notification.consumer.message.buffering.batch.size",
100)).thenReturn(25);
+ return batchConfig;
+ }
+
+ private Configuration buildFailedMsgCacheConfig(int cacheSize) {
+ Configuration cacheConfig = mock(Configuration.class);
+
when(cacheConfig.getInt(NotificationHookConsumer.CONSUMER_FAILEDCACHESIZE_PROPERTY,
1)).thenReturn(cacheSize);
+
when(cacheConfig.getInt(NotificationHookConsumer.CONSUMER_RETRIES_PROPERTY,
3)).thenReturn(3);
+
when(cacheConfig.getBoolean("atlas.notification.create.shell.entity.for.non.existing.ref",
false)).thenReturn(false);
+
when(cacheConfig.getInt("atlas.notification.hook.consumer.buffering.interval",
10)).thenReturn(10);
+
when(cacheConfig.getInt("atlas.notification.hook.consumer.buffering.batch.size",
25)).thenReturn(25);
+
when(cacheConfig.getBoolean("atlas.notification.consumer.create.shell.entity.for.non-existing.ref",
true)).thenReturn(false);
+
when(cacheConfig.getInt("atlas.notification.consumer.message.buffering.interval.seconds",
15)).thenReturn(10);
+
when(cacheConfig.getInt("atlas.notification.consumer.message.buffering.batch.size",
100)).thenReturn(25);
+ return cacheConfig;
+ }
+
+ private Object createEntityProcessor(Configuration configuration) throws
Exception {
+ return new SerialEntityProcessor(
+ configuration,
+ metricsUtil,
+ null,
+ atlasEntityStore,
+ instanceConverter,
+ new EntityCorrelationManager(entityCorrelationStore),
+ typeRegistry,
+ LoggerFactory.getLogger("FAILED"),
+ LoggerFactory.getLogger("LARGE_MESSAGES"),
+ asyncImporter);
+ }
+
+ private Object createEntityProcessorWithPreprocessing() throws Exception {
+ Configuration preprocessingConfig = mock(Configuration.class);
+
when(preprocessingConfig.getBoolean(NotificationHookConsumer.CONSUMER_PREPROCESS_HIVE_TYPES_REMOVE_OWNEDREF_ATTRS,
true)).thenReturn(true);
+
when(preprocessingConfig.getStringArray(NotificationHookConsumer.CONSUMER_PREPROCESS_HIVE_TABLE_IGNORE_PATTERN)).thenReturn(new
String[] {"temp_.*"});
+
when(preprocessingConfig.getInt(NotificationHookConsumer.CONSUMER_RETRIES_PROPERTY,
3)).thenReturn(3);
+
when(preprocessingConfig.getBoolean("atlas.notification.create.shell.entity.for.non.existing.ref",
false)).thenReturn(true);
+
when(preprocessingConfig.getInt("atlas.notification.hook.consumer.buffering.interval",
10)).thenReturn(10);
+
when(preprocessingConfig.getInt("atlas.notification.hook.consumer.buffering.batch.size",
25)).thenReturn(25);
+
when(preprocessingConfig.getBoolean("atlas.notification.consumer.create.shell.entity.for.non-existing.ref",
true)).thenReturn(false);
+
when(preprocessingConfig.getInt("atlas.notification.consumer.message.buffering.interval.seconds",
15)).thenReturn(10);
+
when(preprocessingConfig.getInt("atlas.notification.consumer.message.buffering.batch.size",
100)).thenReturn(25);
+ return createEntityProcessor(preprocessingConfig);
+ }
+
+ private Object createEntityProcessorWithHiveLineageSkip() throws Exception
{
+ Configuration skipConfig = mock(Configuration.class);
+
when(skipConfig.getBoolean(NotificationHookConsumer.CONSUMER_SKIP_HIVE_COLUMN_LINEAGE_HIVE_20633,
false)).thenReturn(true);
+
when(skipConfig.getInt(NotificationHookConsumer.CONSUMER_SKIP_HIVE_COLUMN_LINEAGE_HIVE_20633_INPUTS_THRESHOLD,
15)).thenReturn(1);
+
when(skipConfig.getInt(NotificationHookConsumer.CONSUMER_RETRIES_PROPERTY,
3)).thenReturn(3);
+
when(skipConfig.getBoolean("atlas.notification.create.shell.entity.for.non.existing.ref",
false)).thenReturn(false);
+
when(skipConfig.getInt("atlas.notification.hook.consumer.buffering.interval",
10)).thenReturn(10);
+
when(skipConfig.getInt("atlas.notification.hook.consumer.buffering.batch.size",
25)).thenReturn(25);
+
when(skipConfig.getBoolean("atlas.notification.consumer.create.shell.entity.for.non-existing.ref",
true)).thenReturn(false);
+
when(skipConfig.getInt("atlas.notification.consumer.message.buffering.interval.seconds",
15)).thenReturn(10);
+
when(skipConfig.getInt("atlas.notification.consumer.message.buffering.batch.size",
100)).thenReturn(25);
+ return createEntityProcessor(skipConfig);
+ }
+
+ private Object getEntityProcessor(Object hookConsumer) throws Exception {
+ Field field =
hookConsumer.getClass().getDeclaredField("entityProcessor");
+ field.setAccessible(true);
+ return field.get(hookConsumer);
+ }
+
+ private Object getEntityProcessor(NotificationHookConsumer consumer)
throws Exception {
+ return getEntityProcessor(createHookConsumer(consumer,
mock(NotificationConsumer.class)));
+ }
+
+ private Object createHookConsumerWithEntityProcessor(
+ NotificationHookConsumer notificationHookConsumer,
+ NotificationConsumer<HookNotification> notificationConsumer,
+ Object entityProcessor) throws Exception {
+ Object hookConsumer = createHookConsumer(notificationHookConsumer,
notificationConsumer);
+ Field field =
hookConsumer.getClass().getDeclaredField("entityProcessor");
+ field.setAccessible(true);
+ field.set(hookConsumer, entityProcessor);
+ return hookConsumer;
+ }
+
private Object createHookConsumer(NotificationHookConsumer consumer,
NotificationConsumer<HookNotification> notificationConsumer) throws Exception {
Class<?> hookConsumerClass =
getInnerClass(NotificationHookConsumer.class, "HookConsumer");
return
hookConsumerClass.getDeclaredConstructor(NotificationHookConsumer.class,
NotificationConsumer.class)
