private Map<String, Serde<Object>> getSerdes() {
    Map<String, Serde<Object>> serdeMap = new HashMap<>();
    SerializerConfig serializerConfig = new SerializerConfig(jobConfig);

    // Adding all serdes from factories
    serializerConfig.getSerdeNames()
        .forEach(serdeName -> {
          String serdeClassName = serializerConfig.getSerdeFactoryClass(serdeName)
            .orElseGet(() -> SerializerConfig.getPredefinedSerdeFactoryName(serdeName));
          @SuppressWarnings("unchecked")
          Serde<Object> serde =
              ReflectionUtil.getObj(serdeClassName, SerdeFactory.class).getSerde(serdeName, serializerConfig);
          serdeMap.put(serdeName, serde);
        });

    return serdeMap;
  }
get the max partition number of the changelog stream
samza-core/src/main/java/org/apache/samza/storage/StorageRecovery.java