### Question:
StartProcessProducer extends CamundaBpmProducer { @Override public void process(Exchange exchange) throws Exception { Map<String, Object> processVariables = new HashMap<String, Object>(); if (parameters.containsKey(COPY_MESSAGE_PROPERTIES_PARAMETER)) { processVariables.putAll(exchange.getProperties()); } if (parameters.containsKey(COPY_MESSAGE_HEADERS_PARAMETER)) { processVariables.putAll(exchange.getIn().getHeaders()); } processVariables.putAll(ExchangeUtils.prepareVariables(exchange, parameters)); String processDefinitionKey = this.processDefinitionKey != null ? this.processDefinitionKey : exchange.getIn().getHeader(EXCHANGE_HEADER_PROCESS_DEFINITION_KEY, String.class); ProcessInstance instance = null; if (exchange.getProperties().containsKey(EXCHANGE_HEADER_BUSINESS_KEY)) { instance = runtimeService.startProcessInstanceByKey(processDefinitionKey, exchange.getProperty(EXCHANGE_HEADER_BUSINESS_KEY, String.class), processVariables); exchange.setProperty(EXCHANGE_HEADER_BUSINESS_KEY, instance.getBusinessKey()); } else { instance = runtimeService.startProcessInstanceByKey(processDefinitionKey, processVariables); } exchange.setProperty(EXCHANGE_HEADER_PROCESS_DEFINITION_ID, instance.getProcessDefinitionId()); exchange.setProperty(EXCHANGE_HEADER_PROCESS_INSTANCE_ID, instance.getProcessInstanceId()); exchange.getOut().setBody(instance.getProcessInstanceId()); } StartProcessProducer(CamundaBpmEndpoint endpoint, Map<String, Object> parameters); @Override void close(); @Override void process(Exchange exchange);  }### Answer:
@SuppressWarnings("unchecked") @Test public void startProcessInstanceByKeyShouldBeCalled() throws Exception { ProcessInstance processInstance = mock(ProcessInstance.class); when(processInstance.getProcessInstanceId()).thenReturn("theProcessInstanceId"); when(processInstance.getProcessDefinitionId()).thenReturn("theProcessDefinitionId"); when(runtimeService.startProcessInstanceByKey(eq("aProcessDefinitionKey"), anyMap())).thenReturn( processInstance); CamundaBpmEndpoint endpoint = (CamundaBpmEndpoint) camelContext.getEndpoint( camundaBpmUri("start?" + PROCESS_DEFINITION_KEY_PARAMETER + "=" + "aProcessDefinitionKey")); StartProcessProducer producer = (StartProcessProducer) endpoint.createProducer(); Exchange exchange = new DefaultExchange(camelContext); producer.process(exchange); verify(runtimeService, times(1)).startProcessInstanceByKey(eq("aProcessDefinitionKey"), anyMap()); assertThat(exchange.getProperty(EXCHANGE_HEADER_PROCESS_DEFINITION_ID)).isEqualTo("theProcessDefinitionId"); assertThat(exchange.getProperty(EXCHANGE_HEADER_PROCESS_INSTANCE_ID)).isEqualTo("theProcessInstanceId"); }
@SuppressWarnings("unchecked") @Test public void startProcessInstanceByKeyWithBusinessKeyShouldBeCalled() throws Exception { ProcessInstance processInstance = mock(ProcessInstance.class); when(processInstance.getProcessInstanceId()).thenReturn("theProcessInstanceId"); when(processInstance.getProcessDefinitionId()).thenReturn("theProcessDefinitionId"); when(processInstance.getBusinessKey()).thenReturn("aBusinessKey"); when(runtimeService.startProcessInstanceByKey(eq("aProcessDefinitionKey"), eq("aBusinessKey"), anyMap())).thenReturn(processInstance); CamundaBpmEndpoint endpoint = (CamundaBpmEndpoint) camelContext.getEndpoint( camundaBpmUri("start?" + PROCESS_DEFINITION_KEY_PARAMETER + "=" + "aProcessDefinitionKey")); StartProcessProducer producer = (StartProcessProducer) endpoint.createProducer(); Exchange exchange = new DefaultExchange(camelContext); exchange.setProperty(EXCHANGE_HEADER_BUSINESS_KEY, "aBusinessKey"); producer.process(exchange); verify(runtimeService, times(1)).startProcessInstanceByKey(eq("aProcessDefinitionKey"), eq("aBusinessKey"), anyMap()); assertThat(exchange.getProperty(EXCHANGE_HEADER_PROCESS_DEFINITION_ID)).isEqualTo("theProcessDefinitionId"); assertThat(exchange.getProperty(EXCHANGE_HEADER_PROCESS_INSTANCE_ID)).isEqualTo("theProcessInstanceId"); assertThat(exchange.getProperty(EXCHANGE_HEADER_BUSINESS_KEY)).isEqualTo("aBusinessKey"); }