MicroPipelineFactory { public MicroPipeline instantiatePipeline(final MicroPipelineConfiguration cfg, final ExecutorService executorService) throws RequiredInputMissingException, QueueInitializationFailedException, ComponentInitializationFailedException { if(cfg == null) throw new RequiredInputMissingException("Missing required configuration"); if(StringUtils.isBlank(cfg.getId())) throw new RequiredInputMissingException("Missing required micro pipeline id"); if(cfg.getComponents() == null || cfg.getComponents().isEmpty()) throw new RequiredInputMissingException("Missing required component configurations"); if(cfg.getQueues() == null || cfg.getQueues().isEmpty()) throw new RequiredInputMissingException("Missing required queue configurations"); MetricRegistry.name(StringUtils.lowerCase(StringUtils.trim(this.processingNodeId)), StringUtils.lowerCase(StringUtils.trim(cfg.getId())), "queue", "messages"); final MetricsHandler metricsHandler = new MetricsHandler(); MetricsReporterFactory.attachReporters(metricsHandler, cfg.getMetricsReporter()); MicroPipeline microPipeline = new MicroPipeline(StringUtils.lowerCase(StringUtils.trim(cfg.getId())), cfg); for(final StreamingMessageQueueConfiguration queueConfig : cfg.getQueues()) { String id = StringUtils.lowerCase(StringUtils.trim(queueConfig.getId())); if(microPipeline.hasQueue(id)) { logger.error("queue initialization failed [id="+id+"]. Forcing shutdown of all queues."); microPipeline.shutdown(); throw new QueueInitializationFailedException("Non-unique queue identifier found [id="+id+"]"); } try { StreamingMessageQueue queueInstance = initializeQueue(queueConfig); if(queueConfig.isAttachInsertionCounter()) { final Counter queueInsertionCounter = metricsHandler.counter( MetricRegistry.name( StringUtils.lowerCase(StringUtils.trim(this.processingNodeId)), StringUtils.lowerCase(StringUtils.trim(cfg.getId())), "queue", id, "messages", "in" ), true ); queueInstance.setMessageInsertionCounter(queueInsertionCounter); if(logger.isDebugEnabled()) logger.debug("queue[id="+id+"]: message insertion counter attached"); } if(queueConfig.isAttachRetrievalCounter()) { final Counter queueRetrievalCounter = metricsHandler.counter( MetricRegistry.name( StringUtils.lowerCase(StringUtils.trim(this.processingNodeId)), StringUtils.lowerCase(StringUtils.trim(cfg.getId())), "queue", id, "messages", "out" ), true ); queueInstance.setMessageRetrievalCounter(queueRetrievalCounter); if(logger.isDebugEnabled()) logger.debug("queue[id="+id+"]: message retrieval counter attached"); } microPipeline.addQueue(id, queueInstance); logger.info("queue initialized[id="+id+"]"); } catch(Exception e) { logger.error("queue initialization failed [id="+id+"]. Forcing shutdown of all queues."); microPipeline.shutdown(); throw new QueueInitializationFailedException("Failed to initialize queue [id="+id+"]. Reason: " + e.getMessage(), e); } } final Map<String, MicroPipelineComponent> components = new HashMap<>(); boolean sourceComponentFound = false; boolean emitterComponentFound = false; for(final MicroPipelineComponentConfiguration componentCfg : cfg.getComponents()) { String id = StringUtils.lowerCase(StringUtils.trim(componentCfg.getId())); if(microPipeline.hasComponent(id)) { logger.error("component initialization failed [id="+id+", class="+componentCfg.getName()+", version="+componentCfg.getVersion()+"]. Forcing shutdown of all queues and components."); microPipeline.shutdown(); throw new ComponentInitializationFailedException("Non-unique component identifier found [id="+id+"]"); } try { MicroPipelineComponent component = initializeComponent(componentCfg, microPipeline.getQueues()); if(component.getType() == null) { logger.error("component initialization failed [id="+id+", class="+componentCfg.getName()+", version="+componentCfg.getVersion()+"]. Type missing. Forcing shutdown of all queues and components."); microPipeline.shutdown(); throw new ComponentInitializationFailedException("Failed to initialize component [id="+id+", class="+componentCfg.getName()+", version="+componentCfg.getVersion()+"]. Reason: type missing"); } final StreamingMessageQueue fromQueue = microPipeline.getQueue(StringUtils.lowerCase(StringUtils.trim(componentCfg.getFromQueue()))); final StreamingMessageQueue toQueue = microPipeline.getQueue(StringUtils.lowerCase(StringUtils.trim(componentCfg.getToQueue()))); Counter messageCounter = null; if(componentCfg.isAttachMessageCounter()) { messageCounter = metricsHandler.counter( MetricRegistry.name( StringUtils.lowerCase(StringUtils.trim(this.processingNodeId)), StringUtils.lowerCase(StringUtils.trim(cfg.getId())), "component", id, "messages", "count" ), true ); } switch(component.getType()) { case SOURCE: { SourceRuntimeEnvironment srcEnv = new SourceRuntimeEnvironment(this.processingNodeId, cfg.getId(), (Source)component, toQueue.getProducer()); if(messageCounter != null) srcEnv.setMessageCounter(messageCounter); microPipeline.addSource(id, srcEnv); sourceComponentFound = true; break; } case DIRECT_RESPONSE_OPERATOR: { DirectResponseOperatorRuntimeEnvironment directResponseEnv = new DirectResponseOperatorRuntimeEnvironment(this.processingNodeId, cfg.getId(), (DirectResponseOperator)component, fromQueue.getConsumer(), toQueue.getProducer()); if(componentCfg.isAttachProcessingTimer()) { final Timer messageProcessingTimer = metricsHandler.timer( MetricRegistry.name( StringUtils.lowerCase(StringUtils.trim(this.processingNodeId)), StringUtils.lowerCase(StringUtils.trim(cfg.getId())), "component", id, "messages", "timer" ) ); directResponseEnv.setMessageProcessingTimer(messageProcessingTimer); } if(messageCounter != null) directResponseEnv.setMessageCounter(messageCounter); microPipeline.addOperator(id, directResponseEnv); break; } case DELAYED_RESPONSE_OPERATOR: { DelayedResponseOperatorRuntimeEnvironment delayedResponseEnv = new DelayedResponseOperatorRuntimeEnvironment(this.processingNodeId, cfg.getId(), (DelayedResponseOperator)component, getResponseWaitStrategy(componentCfg), fromQueue.getConsumer(), toQueue.getProducer(), executorService); if(messageCounter != null) delayedResponseEnv.setMessageCounter(messageCounter); microPipeline.addOperator(id, delayedResponseEnv); break; } case EMITTER: { EmitterRuntimeEnvironment emitterEnv = new EmitterRuntimeEnvironment(this.processingNodeId, cfg.getId(), (Emitter)component, fromQueue.getConsumer()); if(componentCfg.isAttachProcessingTimer()) { final Timer messageEmitDurationTimer = metricsHandler.timer( MetricRegistry.name( StringUtils.lowerCase(StringUtils.trim(this.processingNodeId)), StringUtils.lowerCase(StringUtils.trim(cfg.getId())), "component", id, "messages", "emit", "duration" ) ); emitterEnv.setMessageEmitDurationTimer(messageEmitDurationTimer); } if(messageCounter != null) emitterEnv.setMessageCounter(messageCounter); microPipeline.addEmitter(id, emitterEnv); emitterComponentFound = true; break; } } components.put(id, component); } catch(Exception e) { logger.error("component initialization failed [id="+id+", class="+componentCfg.getName()+", version="+componentCfg.getVersion()+"]. Forcing shutdown of all queues and components. Reason: " + e.getMessage(), e); microPipeline.shutdown(); throw new ComponentInitializationFailedException("Failed to initialize component [id="+id+", class="+componentCfg.getName()+", version="+componentCfg.getVersion()+"]. Reason: " + e.getMessage(), e); } } if(!sourceComponentFound) { microPipeline.shutdown(); throw new RequiredInputMissingException("Missing required source component"); } if(!emitterComponentFound) { microPipeline.shutdown(); throw new RequiredInputMissingException("Missing required emitter component"); } microPipeline.attachComponentMetricsHandler(metricsHandler); for(String sourceId : microPipeline.getSources().keySet()) { executorService.submit(microPipeline.getSources().get(sourceId)); if(logger.isDebugEnabled()) logger.debug("Started runtime environment for source [id="+sourceId+"]"); } for(String directResponseOperatorId : microPipeline.getDirectResponseOperators().keySet()) { executorService.submit(microPipeline.getDirectResponseOperators().get(directResponseOperatorId)); if(logger.isDebugEnabled()) logger.debug("Started runtime environment for direct response operator [id="+directResponseOperatorId+"]"); } for(String delayedResponseOperatorId : microPipeline.getDelayedResponseOperators().keySet()) { executorService.submit(microPipeline.getDelayedResponseOperators().get(delayedResponseOperatorId)); if(logger.isDebugEnabled()) logger.debug("Started runtime environment for delayed response operator [id="+delayedResponseOperatorId+"]"); } for(String emitterId : microPipeline.getEmitters().keySet()) { executorService.submit(microPipeline.getEmitters().get(emitterId)); if(logger.isDebugEnabled()) logger.debug("Started runtime environment for emitter [id="+emitterId+"]"); } if(logger.isDebugEnabled()) logger.debug("Started stats collector"); return microPipeline; } MicroPipelineFactory(final String processingNodeId, final ComponentRepository componentRepository); MicroPipeline instantiatePipeline(final MicroPipelineConfiguration cfg, final ExecutorService executorService);  }
@Test public void testInstantiatePipeline_withNullProcessingNodeId() throws QueueInitializationFailedException, ComponentInitializationFailedException { try { new MicroPipelineFactory(null, Mockito.mock(ComponentRepository.class)).instantiatePipeline(new MicroPipelineConfiguration(), executorService); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInstantiatePipeline_withNullInput() throws QueueInitializationFailedException, ComponentInitializationFailedException { try { new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).instantiatePipeline(null, executorService); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInstantiatePipeline_withEmptyInput() throws QueueInitializationFailedException, ComponentInitializationFailedException { try { new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).instantiatePipeline(new MicroPipelineConfiguration(), executorService); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInstantiatePipeline_withEmptyId() throws QueueInitializationFailedException, ComponentInitializationFailedException { MicroPipelineComponentConfiguration componentCfg = Mockito.mock(MicroPipelineComponentConfiguration.class); StreamingMessageQueueConfiguration queueCfg = Mockito.mock(StreamingMessageQueueConfiguration.class); try { MicroPipelineConfiguration cfg = new MicroPipelineConfiguration(); cfg.setId(""); cfg.getComponents().add(componentCfg); cfg.getQueues().add(queueCfg); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).instantiatePipeline(cfg, executorService); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInstantiatePipeline_withEmptyComponentCfg() throws QueueInitializationFailedException, ComponentInitializationFailedException { StreamingMessageQueueConfiguration queueCfg = Mockito.mock(StreamingMessageQueueConfiguration.class); try { MicroPipelineConfiguration cfg = new MicroPipelineConfiguration(); cfg.setId("test"); cfg.getQueues().add(queueCfg); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).instantiatePipeline(cfg, executorService); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInstantiatePipeline_withEmptyQueueCfg() throws QueueInitializationFailedException, ComponentInitializationFailedException { MicroPipelineComponentConfiguration componentCfg = Mockito.mock(MicroPipelineComponentConfiguration.class); try { MicroPipelineConfiguration cfg = new MicroPipelineConfiguration(); cfg.setId("test"); cfg.getComponents().add(componentCfg); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).instantiatePipeline(cfg, executorService); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInstantiatePipeline_withNonUniqueQueueId() throws RequiredInputMissingException, ComponentInitializationFailedException { MicroPipelineComponentConfiguration componentCfg = Mockito.mock(MicroPipelineComponentConfiguration.class); Properties queueProps = new Properties(); StreamingMessageQueueConfiguration validQueueCfg = new StreamingMessageQueueConfiguration(); validQueueCfg.setId("testInstantiatePipeline_withNonUniqueQueueId"); validQueueCfg.setProperties(queueProps); StreamingMessageQueueConfiguration invalidQueueCfg = new StreamingMessageQueueConfiguration(); invalidQueueCfg.setId("testInstantiatePipeline_withNonUniqueQueueId"); invalidQueueCfg.setProperties(queueProps); try { MicroPipelineConfiguration cfg = new MicroPipelineConfiguration(); cfg.setId("test"); cfg.getComponents().add(componentCfg); cfg.getQueues().add(validQueueCfg); cfg.getQueues().add(invalidQueueCfg); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).instantiatePipeline(cfg, executorService); Assert.fail("Missing required input"); } catch(QueueInitializationFailedException e) { } }
@Test public void testInstantiatePipeline_withInvalidComponentConfiguration() throws RequiredInputMissingException, QueueInitializationFailedException, ComponentInstantiationFailedException, UnknownComponentException { StreamingMessageQueueConfiguration validQueueCfg = new StreamingMessageQueueConfiguration(); validQueueCfg.setId("queue-1"); MicroPipelineComponentConfiguration component1Cfg = new MicroPipelineComponentConfiguration(); component1Cfg.setId("test-id"); component1Cfg.setSettings(new Properties()); component1Cfg.setType(MicroPipelineComponentType.SOURCE); component1Cfg.setName("testInitializeComponent_source_withReferenceToUnknownQueue"); component1Cfg.setVersion("0.0.1"); component1Cfg.setToQueue("queue-1"); MicroPipelineComponentConfiguration component2Cfg = new MicroPipelineComponentConfiguration(); component2Cfg.setId("test-id-2"); component2Cfg.setSettings(new Properties()); component2Cfg.setType(MicroPipelineComponentType.SOURCE); component2Cfg.setName("testInitializeComponent_source_withReferenceToUnknownQueue-2"); component2Cfg.setVersion("0.0.1"); component2Cfg.setToQueue("queue-1"); MicroPipelineConfiguration pipelineCfg = new MicroPipelineConfiguration(); pipelineCfg.setId("test-id"); pipelineCfg.getQueues().add(validQueueCfg); pipelineCfg.getComponents().add(component2Cfg); pipelineCfg.getComponents().add(component1Cfg); MicroPipelineComponent mockComponent2 = Mockito.mock(MicroPipelineComponent.class); Mockito.when(mockComponent2.getId()).thenReturn("mock-component-2"); ComponentRepository repo = Mockito.mock(ComponentRepository.class); Mockito.when(repo.newInstance(component1Cfg.getId(), component1Cfg.getName(), component1Cfg.getVersion(), component1Cfg.getSettings())).thenThrow(new UnknownComponentException("Unknown component class")); Mockito.when(repo.newInstance(component2Cfg.getId(), component2Cfg.getName(), component2Cfg.getVersion(), component2Cfg.getSettings())).thenReturn(mockComponent2); try { new MicroPipelineFactory("id", repo).instantiatePipeline(pipelineCfg, executorService); Assert.fail("Missing required input"); } catch(ComponentInitializationFailedException e) { } }
@Test public void testInstantiatePipeline_withComponentConfigurationTwice() throws RequiredInputMissingException, QueueInitializationFailedException, ComponentInstantiationFailedException, UnknownComponentException { StreamingMessageQueueConfiguration validQueueCfg = new StreamingMessageQueueConfiguration(); validQueueCfg.setId("queue-1"); MicroPipelineComponentConfiguration component1Cfg = new MicroPipelineComponentConfiguration(); component1Cfg.setId("test-id"); component1Cfg.setSettings(new Properties()); component1Cfg.setType(MicroPipelineComponentType.SOURCE); component1Cfg.setName("testInitializeComponent_source_withReferenceToUnknownQueue"); component1Cfg.setVersion("0.0.1"); component1Cfg.setToQueue("queue-1"); MicroPipelineConfiguration pipelineCfg = new MicroPipelineConfiguration(); pipelineCfg.setId("test-id"); pipelineCfg.getQueues().add(validQueueCfg); pipelineCfg.getComponents().add(component1Cfg); pipelineCfg.getComponents().add(component1Cfg); MicroPipelineComponent mockComponent1 = Mockito.mock(MicroPipelineComponent.class); Mockito.when(mockComponent1.getId()).thenReturn("mock-component-1"); Mockito.when(mockComponent1.getType()).thenReturn(MicroPipelineComponentType.SOURCE); ComponentRepository repo = Mockito.mock(ComponentRepository.class); Mockito.when(repo.newInstance(component1Cfg.getId(), component1Cfg.getName(), component1Cfg.getVersion(), component1Cfg.getSettings())).thenReturn(mockComponent1); try { new MicroPipelineFactory("id", repo).instantiatePipeline(pipelineCfg, executorService); Assert.fail("Missing required input"); } catch(ComponentInitializationFailedException e) { } }
@Test public void testInstantiatePipeline_withComponentTypeNull() throws RequiredInputMissingException, QueueInitializationFailedException, ComponentInstantiationFailedException, UnknownComponentException { StreamingMessageQueueConfiguration validQueueCfg = new StreamingMessageQueueConfiguration(); validQueueCfg.setId("queue-1"); MicroPipelineComponentConfiguration component1Cfg = new MicroPipelineComponentConfiguration(); component1Cfg.setId("test-id"); component1Cfg.setSettings(new Properties()); component1Cfg.setType(MicroPipelineComponentType.SOURCE); component1Cfg.setName("testInitializeComponent_source_withReferenceToUnknownQueue"); component1Cfg.setVersion("0.0.1"); component1Cfg.setToQueue("queue-1"); MicroPipelineConfiguration pipelineCfg = new MicroPipelineConfiguration(); pipelineCfg.setId("test-id"); pipelineCfg.getQueues().add(validQueueCfg); pipelineCfg.getComponents().add(component1Cfg); MicroPipelineComponent mockComponent1 = Mockito.mock(MicroPipelineComponent.class); Mockito.when(mockComponent1.getId()).thenReturn("mock-component-1"); Mockito.when(mockComponent1.getType()).thenReturn(null); ComponentRepository repo = Mockito.mock(ComponentRepository.class); Mockito.when(repo.newInstance(component1Cfg.getId(), component1Cfg.getName(), component1Cfg.getVersion(), component1Cfg.getSettings())).thenReturn(mockComponent1); try { new MicroPipelineFactory("id", repo).instantiatePipeline(pipelineCfg, executorService); Assert.fail("Missing required input"); } catch(ComponentInitializationFailedException e) { } }