final void addTableDescriptor(TableDescriptor tableDescriptor) {
    String tableId = tableDescriptor.getTableId();
    Preconditions.checkState(StringUtils.isNotBlank(tableId) && ID_PATTERN.matcher(tableId).matches(),
        String.format("tableId: %s must confirm to pattern: %s", tableId, ID_PATTERN.toString()));
    Preconditions.checkState(!tableDescriptors.containsKey(tableId)
        || tableDescriptors.get(tableId) == tableDescriptor,
        String.format("Cannot add multiple table descriptors with the same tableId: %s", tableId));

    if (tableDescriptor instanceof HybridTableDescriptor) {
      List<? extends TableDescriptor> tableDescs =
          ((HybridTableDescriptor) tableDescriptor).getTableDescriptors();
      tableDescs.forEach(td -> addTableDescriptor(td));
    }

    tableDescriptors.put(tableId, tableDescriptor);
  }
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