KV<Serde, Serde> getOrCreateTableSerdes(String tableId, KVSerde kvSerde) {
    Serde keySerde, valueSerde;
    keySerde = kvSerde.getKeySerde();
    valueSerde = kvSerde.getValueSerde();

    if (!tableSerdes.containsKey(tableId)) {
      tableSerdes.put(tableId, KV.of(keySerde, valueSerde));
      return tableSerdes.get(tableId);
    }

    KV<Serde, Serde> currentSerdePair = tableSerdes.get(tableId);
    if (!currentSerdePair.getKey().equals(keySerde) || !currentSerdePair.getValue().equals(valueSerde)) {
      throw new IllegalArgumentException(String.format("Serde for table %s is already defined. Cannot change it to "
          + "different serdes.", tableId));
    }
    return streamSerdes.get(tableId);
  }
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