final void addInputDescriptor(InputDescriptor inputDescriptor) {
    String streamId = inputDescriptor.getStreamId();
    Preconditions.checkState(!inputDescriptors.containsKey(streamId)
            || inputDescriptors.get(streamId) == inputDescriptor,
        String.format("Cannot add multiple input descriptors with the same streamId: %s", streamId));
    inputDescriptors.put(streamId, inputDescriptor);
    addSystemDescriptor(inputDescriptor.getSystemDescriptor());
  }
Get the corresponding {@link KVSerde} for the input {@code inputStreamId}

@param tableId id of the table
@return the {@link KVSerde} for the stream. null if the serde is not defined or {@code streamId} does not exist
samza-core/src/main/java/org/apache/samza/application/descriptors/ApplicationDescriptorImpl.java