StormEventCorrelationInjector { public StreamlineEvent injectCorrelationInformation(List<Object> tuple, List<Tuple> parentTuples, String componentName) { EventCorrelationInjector eventCorrelationInjector = new EventCorrelationInjector(); return eventCorrelationInjector.injectCorrelationInformation( getStreamlineEventFromValues(tuple), parentTuples.stream().map(this::getStreamlineEventFromTuple).collect(Collectors.toList()), StormTopologyUtil.extractStreamlineComponentName(componentName)); }  StreamlineEvent injectCorrelationInformation(List<Object> tuple, List<Tuple> parentTuples, String componentName);  }
@Test public void tupleArgsTestNoParent() throws Exception { StreamlineEvent injectedEvent = sut.injectCorrelationInformation( new Values(INPUT_STREAMLINE_EVENT), Collections.emptyList(), TEST_COMPONENT_NAME); StreamlineEventTestUtil.assertEventIsProperlyCopied(injectedEvent, INPUT_STREAMLINE_EVENT); assertEquals(Collections.emptySet(), EventCorrelationInjector.getRootIds(injectedEvent)); assertEquals(Collections.emptySet(), EventCorrelationInjector.getParentIds(injectedEvent)); assertEquals(StormTopologyUtil.extractStreamlineComponentName(TEST_COMPONENT_NAME), EventCorrelationInjector.getSourceComponentName(injectedEvent)); }
@Test public void tupleArgsTestWithAParentWhichIsRootEvent() throws Exception { StreamlineEvent parentEvent = copyEventWithNewID(); parentEvent = eventCorrelationInjector.injectCorrelationInformation(parentEvent, Collections.emptyList(), TEST_COMPONENT_NAME); int parentTaskId = 1; String parentComponentName = "1-parentComponent"; new Expectations() {{ mockedTopologyContext.getComponentId(parentTaskId); result = parentComponentName; mockedTopologyContext.getComponentOutputFields(parentComponentName, INPUT_STREAMLINE_EVENT.getSourceStream()); result = new Fields(StreamlineEvent.STREAMLINE_EVENT); }}; Tuple parentTuple = new TupleImpl(mockedTopologyContext, new Values(parentEvent), parentTaskId, INPUT_STREAMLINE_EVENT.getSourceStream()); StreamlineEvent injectedEvent = sut.injectCorrelationInformation( new Values(INPUT_STREAMLINE_EVENT), Collections.singletonList(parentTuple), TEST_COMPONENT_NAME); StreamlineEventTestUtil.assertEventIsProperlyCopied(injectedEvent, INPUT_STREAMLINE_EVENT); assertEquals(Collections.singleton(parentEvent.getId()), EventCorrelationInjector.getRootIds(injectedEvent)); assertEquals(Collections.singleton(parentEvent.getId()), EventCorrelationInjector.getParentIds(injectedEvent)); assertEquals(StormTopologyUtil.extractStreamlineComponentName(TEST_COMPONENT_NAME), EventCorrelationInjector.getSourceComponentName(injectedEvent)); }
@Test public void tupleArgsTestWithTwoParentsWhichAreRootEvents() throws Exception { int parent1TaskId = 1; String parent1ComponentName = "1-parentComponent"; int parent2TaskId = 2; String parent2ComponentName = "2-parentComponent2"; new Expectations() {{ mockedTopologyContext.getComponentId(parent1TaskId); result = parent1ComponentName; mockedTopologyContext.getComponentId(parent2TaskId); result = parent2ComponentName; mockedTopologyContext.getComponentOutputFields(parent1ComponentName, INPUT_STREAMLINE_EVENT.getSourceStream()); result = new Fields(StreamlineEvent.STREAMLINE_EVENT); mockedTopologyContext.getComponentOutputFields(parent2ComponentName, INPUT_STREAMLINE_EVENT.getSourceStream()); result = new Fields(StreamlineEvent.STREAMLINE_EVENT); }}; StreamlineEvent parentEvent1 = copyEventWithNewID(); parentEvent1 = eventCorrelationInjector.injectCorrelationInformation(parentEvent1, Collections.emptyList(), TEST_COMPONENT_NAME); Tuple parentTuple1 = new TupleImpl(mockedTopologyContext, new Values(parentEvent1), parent1TaskId, INPUT_STREAMLINE_EVENT.getSourceStream()); StreamlineEvent parentEvent2 = copyEventWithNewID(); parentEvent2 = eventCorrelationInjector.injectCorrelationInformation(parentEvent2, Collections.emptyList(), TEST_COMPONENT_NAME); Tuple parentTuple2 = new TupleImpl(mockedTopologyContext, new Values(parentEvent2), parent2TaskId, INPUT_STREAMLINE_EVENT.getSourceStream()); StreamlineEvent injectedEvent = sut.injectCorrelationInformation( new Values(INPUT_STREAMLINE_EVENT), Lists.newArrayList(parentTuple1, parentTuple2), TEST_COMPONENT_NAME); StreamlineEventTestUtil.assertEventIsProperlyCopied(injectedEvent, INPUT_STREAMLINE_EVENT); assertEquals(Sets.newHashSet(parentEvent1.getId(), parentEvent2.getId()), EventCorrelationInjector.getRootIds(injectedEvent)); assertEquals(Sets.newHashSet(parentEvent1.getId(), parentEvent2.getId()), EventCorrelationInjector.getParentIds(injectedEvent)); assertEquals(StormTopologyUtil.extractStreamlineComponentName(TEST_COMPONENT_NAME), EventCorrelationInjector.getSourceComponentName(injectedEvent)); }
@Test public void tupleArgsTestWithTwoParentsWhichOneIsRootEventAndAnotherOneIsNonRootEvent() throws Exception { int parent1TaskId = 1; String parent1ComponentName = "1-parentComponent"; int parent2TaskId = 2; String parent2ComponentName = "2-parentComponent2"; new Expectations() {{ mockedTopologyContext.getComponentId(parent1TaskId); result = parent1ComponentName; mockedTopologyContext.getComponentId(parent2TaskId); result = parent2ComponentName; mockedTopologyContext.getComponentOutputFields(parent1ComponentName, INPUT_STREAMLINE_EVENT.getSourceStream()); result = new Fields(StreamlineEvent.STREAMLINE_EVENT); mockedTopologyContext.getComponentOutputFields(parent2ComponentName, INPUT_STREAMLINE_EVENT.getSourceStream()); result = new Fields(StreamlineEvent.STREAMLINE_EVENT); }}; StreamlineEvent parentEvent1 = copyEventWithNewID(); parentEvent1 = eventCorrelationInjector.injectCorrelationInformation(parentEvent1, Collections.emptyList(), TEST_COMPONENT_NAME); Tuple parentTuple1 = new TupleImpl(mockedTopologyContext, new Values(parentEvent1), parent1TaskId, INPUT_STREAMLINE_EVENT.getSourceStream()); StreamlineEvent parentOfParentEvent2 = copyEventWithNewID(); parentOfParentEvent2 = eventCorrelationInjector.injectCorrelationInformation(parentOfParentEvent2, Collections.emptyList(), TEST_COMPONENT_NAME); StreamlineEvent parentEvent2 = copyEventWithNewID(); parentEvent2 = eventCorrelationInjector.injectCorrelationInformation(parentEvent2, Collections.singletonList(parentOfParentEvent2), TEST_COMPONENT_NAME); Tuple parentTuple2 = new TupleImpl(mockedTopologyContext, new Values(parentEvent2), parent2TaskId, INPUT_STREAMLINE_EVENT.getSourceStream()); StreamlineEvent injectedEvent = sut.injectCorrelationInformation( new Values(INPUT_STREAMLINE_EVENT), Lists.newArrayList(parentTuple1, parentTuple2), TEST_COMPONENT_NAME); StreamlineEventTestUtil.assertEventIsProperlyCopied(injectedEvent, INPUT_STREAMLINE_EVENT); assertEquals(Sets.newHashSet(parentEvent1.getId(), parentOfParentEvent2.getId()), EventCorrelationInjector.getRootIds(injectedEvent)); assertEquals(Sets.newHashSet(parentEvent1.getId(), parentEvent2.getId()), EventCorrelationInjector.getParentIds(injectedEvent)); assertEquals(StormTopologyUtil.extractStreamlineComponentName(TEST_COMPONENT_NAME), EventCorrelationInjector.getSourceComponentName(injectedEvent)); }