private void setAwsClientConfigs(Config config, ClientConfiguration clientConfig) {
    for (Entry<String, String> entry : config.entrySet()) {
      boolean found = false;
      String key = entry.getKey();
      String value = entry.getValue();
      if (StringUtils.isEmpty(value)) {
        continue;
      }
      for (Method method : ClientConfiguration.class.getMethods()) {
        // For each property invoke the corresponding setter, if it exists
        if (method.getName().equals("set" + key)) {
          found = true;
          Class<?> type = method.getParameterTypes()[0];
          try {
            if (type == long.class) {
              method.invoke(clientConfig, Long.valueOf(value));
            } else if (type == int.class) {
              method.invoke(clientConfig, Integer.valueOf(value));
            } else if (type == boolean.class) {
              method.invoke(clientConfig, Boolean.valueOf(value));
            } else if (type == String.class) {
              method.invoke(clientConfig, value);
            }
            LOG.info("Loaded property " + key + " = " + value);
            break;
          } catch (Exception e) {
            throw new IllegalArgumentException(
                String.format("Error trying to set field %s with the value '%s'", key, value), e);
          }
        }
      }
      if (!found) {
        LOG.warn("Property " + key + " ignored as there is no corresponding set method");
      }
    }
  }
Get the appropriate CredentialProvider for a given system stream.
@param system name of the system
@param stream name of the stream
@return AWSCredentialsProvider
samza-aws/src/main/java/org/apache/samza/system/kinesis/KinesisConfig.java