KV<Serde, Serde> getOrCreateStreamSerdes(String streamId, Serde serde) {
    Serde keySerde, valueSerde;

    KV<Serde, Serde> currentSerdePair = streamSerdes.get(streamId);

    if (serde instanceof KVSerde) {
      keySerde = ((KVSerde) serde).getKeySerde();
      valueSerde = ((KVSerde) serde).getValueSerde();
    } else {
      keySerde = new NoOpSerde();
      valueSerde = serde;
    }

    if (currentSerdePair == null) {
      if (keySerde instanceof NoOpSerde) {
        LOGGER.info("Using NoOpSerde as the key serde for stream " + streamId +
            ". Keys will not be (de)serialized");
      }
      if (valueSerde instanceof NoOpSerde) {
        LOGGER.info("Using NoOpSerde as the value serde for stream " + streamId +
            ". Values will not be (de)serialized");
      }
      streamSerdes.put(streamId, KV.of(keySerde, valueSerde));
    } else if (!currentSerdePair.getKey().getClass().equals(keySerde.getClass())
        || !currentSerdePair.getValue().getClass().equals(valueSerde.getClass())) {
      throw new IllegalArgumentException(String.format("Serde for streamId: %s is already defined. Cannot change it to "
          + "different serdes.", streamId));
    } else {
      LOGGER.warn("Using previously defined serde for streamId: " + streamId + ".");
    }
    return streamSerdes.get(streamId);
  }
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