StreamApplicationDescriptorImpl extends ApplicationDescriptorImpl<StreamApplicationDescriptor> implements StreamApplicationDescriptor { @VisibleForTesting public <M> IntermediateMessageStreamImpl<M> getIntermediateStream(String streamId, Serde<M> serde, boolean isBroadcast) { Preconditions.checkNotNull(serde, "serde must not be null for intermediate stream: " + streamId); Preconditions.checkState(!inputOperators.containsKey(streamId) && !outputStreams.containsKey(streamId), "getIntermediateStream must not be called multiple times with the same streamId: " + streamId); if (isBroadcast) { intermediateBroadcastStreamIds.add(streamId); } boolean isKeyed = serde instanceof KVSerde; KV<Serde, Serde> kvSerdes = getOrCreateStreamSerdes(streamId, serde); InputTransformer transformer = (InputTransformer) getDefaultSystemDescriptor() .flatMap(SystemDescriptor::getTransformer).orElse(null); InputOperatorSpec inputOperatorSpec = OperatorSpecs.createInputOperatorSpec(streamId, kvSerdes.getKey(), kvSerdes.getValue(), transformer, isKeyed, this.getNextOpId(OpCode.INPUT, null)); inputOperators.put(streamId, inputOperatorSpec); outputStreams.put(streamId, new OutputStreamImpl(streamId, kvSerdes.getKey(), kvSerdes.getValue(), isKeyed)); return new IntermediateMessageStreamImpl<>(this, inputOperators.get(streamId), outputStreams.get(streamId)); } StreamApplicationDescriptorImpl(StreamApplication userApp, Config config); @Override MessageStream<M> getInputStream(InputDescriptor<M, ?> inputDescriptor); @Override OutputStream<M> getOutputStream(OutputDescriptor<M, ?> outputDescriptor); @Override Table<KV<K, V>> getTable(TableDescriptor<K, V, ?> tableDescriptor); @Override Set<String> getInputStreamIds(); @Override Set<String> getOutputStreamIds(); @Override Set<String> getIntermediateBroadcastStreamIds(); Map<String, InputOperatorSpec> getInputOperators(); Map<String, OutputStreamImpl> getOutputStreams(); OperatorSpecGraph getOperatorSpecGraph(); String getNextOpId(OpCode opCode, String userDefinedId); String getNextOpId(OpCode opCode); @VisibleForTesting IntermediateMessageStreamImpl<M> getIntermediateStream(String streamId, Serde<M> serde, boolean isBroadcast);  }
@Test(expected = IllegalStateException.class) public void testSetDefaultSystemDescriptorAfterGettingIntermediateStream() { String streamId = "test-stream-1"; StreamApplicationDescriptorImpl streamAppDesc = new StreamApplicationDescriptorImpl(appDesc -> { }, getConfig()); streamAppDesc.getIntermediateStream(streamId, mock(Serde.class), false); streamAppDesc.withDefaultSystem(mock(SystemDescriptor.class)); }
@Test(expected = NullPointerException.class) public void testGetIntermediateStreamWithNoSerde() { Config mockConfig = getConfig(); String streamId = "streamId"; StreamApplicationDescriptorImpl streamAppDesc = new StreamApplicationDescriptorImpl(appDesc -> { }, mockConfig); IntermediateMessageStreamImpl<TestMessageEnvelope> intermediateStreamImpl = streamAppDesc.getIntermediateStream(streamId, null, false); }
@Test(expected = IllegalStateException.class) public void testGetSameIntermediateStreamTwice() { StreamApplicationDescriptorImpl streamAppDesc = new StreamApplicationDescriptorImpl(appDesc -> { }, getConfig()); streamAppDesc.getIntermediateStream("test-stream-1", mock(Serde.class), false); streamAppDesc.getIntermediateStream("test-stream-1", mock(Serde.class), false); }