@SuppressWarnings("unchecked") @Test (expected = StreamsException.class) public void shouldThrowStreamsExceptionIfExceptionCaughtDuringClose() throws Exception { final ProcessorNode node = new ProcessorNode("name", new ExceptionalProcessor(), Collections.emptySet()); node.close(); }
public void close() { try { nodeMetrics.metrics.measureLatencyNs(time, closeDelegate, nodeMetrics.nodeDestructionSensor); nodeMetrics.removeAllSensors(); } catch (Exception e) { throw new StreamsException(String.format("failed to close processor %s", name), e); } }
ProcessorNode { public void close() { try { nodeMetrics.metrics.measureLatencyNs(time, closeDelegate, nodeMetrics.nodeDestructionSensor); nodeMetrics.removeAllSensors(); } catch (Exception e) { throw new StreamsException(String.format("failed to close processor %s", name), e); } } }
ProcessorNode { public void close() { try { nodeMetrics.metrics.measureLatencyNs(time, closeDelegate, nodeMetrics.nodeDestructionSensor); nodeMetrics.removeAllSensors(); } catch (Exception e) { throw new StreamsException(String.format("failed to close processor %s", name), e); } } ProcessorNode(String name);  ProcessorNode(String name, Processor<K, V> processor, Set<String> stateStores); }
ProcessorNode { public void close() { try { nodeMetrics.metrics.measureLatencyNs(time, closeDelegate, nodeMetrics.nodeDestructionSensor); nodeMetrics.removeAllSensors(); } catch (Exception e) { throw new StreamsException(String.format("failed to close processor %s", name), e); } } ProcessorNode(String name);  ProcessorNode(String name, Processor<K, V> processor, Set<String> stateStores); final String name(); final Processor<K, V> processor(); final List<ProcessorNode<?, ?>> children(); void addChild(ProcessorNode<?, ?> child); void init(ProcessorContext context); void close(); void process(final K key, final V value); void punctuate(long timestamp); @Override String toString(); String toString(String indent); }
ProcessorNode { public void close() { try { nodeMetrics.metrics.measureLatencyNs(time, closeDelegate, nodeMetrics.nodeDestructionSensor); nodeMetrics.removeAllSensors(); } catch (Exception e) { throw new StreamsException(String.format("failed to close processor %s", name), e); } } ProcessorNode(String name);  ProcessorNode(String name, Processor<K, V> processor, Set<String> stateStores); final String name(); final Processor<K, V> processor(); final List<ProcessorNode<?, ?>> children(); void addChild(ProcessorNode<?, ?> child); void init(ProcessorContext context); void close(); void process(final K key, final V value); void punctuate(long timestamp); @Override String toString(); String toString(String indent); final Set<String> stateStores; }