StreamApplicationDescriptorImpl extends ApplicationDescriptorImpl<StreamApplicationDescriptor> implements StreamApplicationDescriptor { @Override public <K, V> Table<KV<K, V>> getTable(TableDescriptor<K, V, ?> tableDescriptor) { addTableDescriptor(tableDescriptor); if (tableDescriptor instanceof LocalTableDescriptor) { LocalTableDescriptor localTableDescriptor = (LocalTableDescriptor) tableDescriptor; getOrCreateTableSerdes(localTableDescriptor.getTableId(), localTableDescriptor.getSerde()); } return new TableImpl(tableDescriptor); } 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 public void testGetTable() throws Exception { Config mockConfig = getConfig(); String tableId = "t1"; BaseTableDescriptor mockTableDescriptor = mock(BaseTableDescriptor.class); when(mockTableDescriptor.getTableId()).thenReturn(tableId); AtomicReference<TableImpl> table = new AtomicReference<>(); StreamApplicationDescriptorImpl streamAppDesc = new StreamApplicationDescriptorImpl(appDesc -> { table.set((TableImpl) appDesc.getTable(mockTableDescriptor)); }, mockConfig); assertEquals(tableId, table.get().getTableId()); }
@Test(expected = IllegalStateException.class) public void testGetTableWithBadId() { Config mockConfig = getConfig(); new StreamApplicationDescriptorImpl(appDesc -> { BaseTableDescriptor mockTableDescriptor = mock(BaseTableDescriptor.class); when(mockTableDescriptor.getTableId()).thenReturn("my.table"); appDesc.getTable(mockTableDescriptor); }, mockConfig); }