@Experimental
public interface CustomMetric {
    enum AggregationMethod {
        MIN,
        MAX,
        SUM,
        AVG
    }

    CustomMetric addAggregationMethod(AggregationMethod aggregationMethod);
}

@Experimental
public interface CustomInitializationMetrics {
    CustomMetric addMetric(String name, long value);
    CustomMetric addMetric(String name, double value);
}

// intendend usage:
customInitializationMetrics.addMetric("DownloadTime", value)
    .addAggregationMethod(AggregationMethod.MAX)
    .addAggregationMethod(AggregationMethod.SUM)

@PublicEvolving
public interface StateBackend {
  (...)
  
  @PublicEvolving
  pubic interface KeyedStateBackendParameters<K> {
    // already existing parameters of createKeyedStateBackend() method
    Environment getEnv();
    JobID getJobID();
    String getOperatorIdentifier();
    TypeSerializer<K> getKeySerializer();
    int getNumberOfKeyGroups();
    KeyGroupRange getKeyGroupRange();
    TaskKvStateRegistry getKvStateRegistry();
    TtlTimeProvider getTtlTimeProvider();
    @Nonnull Collection<KeyedStateHandle> getStateHandles();
    CloseableRegistry getCancelStreamRegistry();
    double getManagedMemoryFraction();
    
    // newly added parameter
    CustomInitializationMetrics getCustomInitializationMetrics();
  }
  
  @PublicEvolving
  public interface OperatorStateBackendParameters {
    // already existing parameters of createOperatorStateBackend() method
    Environment getEnv();
    String getOperatorIdentifier();
    @Nonnull Collection<OperatorStateHandle> getStateHandles();
    CloseableRegistry getCancelStreamRegistry();
    
    // newly added parameter
    CustomInitializationMetrics getCustomInitializationMetrics();
  }
}

@PublicEvolving
public interface StateBackend {
  (...)

  <K> CheckpointableKeyedStateBackend<K> createKeyedStateBackend(
         KeyedStateBackendParameters<K> parameters);
      
  OperatorStateBackend createOperatorStateBackend(
          OperatorStateBackendParameters parameters);
}