@Test public void testThreadingModelInProducerConfiguration() { Properties testConfig = new Properties(); testConfig.setProperty(AWSConfigConstants.AWS_REGION, "us-east-1"); KinesisProducerConfiguration kpc = KinesisConfigUtil.getValidatedProducerConfiguration(testConfig); assertEquals(KinesisProducerConfiguration.ThreadingModel.POOLED, kpc.getThreadingModel()); testConfig.setProperty(KinesisConfigUtil.THREADING_MODEL, "PER_REQUEST"); kpc = KinesisConfigUtil.getValidatedProducerConfiguration(testConfig); assertEquals(KinesisProducerConfiguration.ThreadingModel.PER_REQUEST, kpc.getThreadingModel()); }
public static KinesisProducerConfiguration getValidatedProducerConfiguration(Properties config) { checkNotNull(config, "config can not be null"); validateAwsConfiguration(config); if (!config.containsKey(AWSConfigConstants.AWS_REGION)) { throw new IllegalArgumentException(String.format("For FlinkKinesisProducer AWS region ('%s') must be set in the config.", AWSConfigConstants.AWS_REGION)); } KinesisProducerConfiguration kpc = KinesisProducerConfiguration.fromProperties(config); kpc.setRegion(config.getProperty(AWSConfigConstants.AWS_REGION)); kpc.setCredentialsProvider(AWSUtil.getCredentialsProvider(config)); kpc.setCredentialsRefreshDelay(100); if (!config.containsKey(RATE_LIMIT)) { kpc.setRateLimit(DEFAULT_RATE_LIMIT); } if (!config.containsKey(THREADING_MODEL)) { kpc.setThreadingModel(DEFAULT_THREADING_MODEL); } if (!config.containsKey(THREAD_POOL_SIZE)) { kpc.setThreadPoolSize(DEFAULT_THREAD_POOL_SIZE); } return kpc; }
KinesisConfigUtil { public static KinesisProducerConfiguration getValidatedProducerConfiguration(Properties config) { checkNotNull(config, "config can not be null"); validateAwsConfiguration(config); if (!config.containsKey(AWSConfigConstants.AWS_REGION)) { throw new IllegalArgumentException(String.format("For FlinkKinesisProducer AWS region ('%s') must be set in the config.", AWSConfigConstants.AWS_REGION)); } KinesisProducerConfiguration kpc = KinesisProducerConfiguration.fromProperties(config); kpc.setRegion(config.getProperty(AWSConfigConstants.AWS_REGION)); kpc.setCredentialsProvider(AWSUtil.getCredentialsProvider(config)); kpc.setCredentialsRefreshDelay(100); if (!config.containsKey(RATE_LIMIT)) { kpc.setRateLimit(DEFAULT_RATE_LIMIT); } if (!config.containsKey(THREADING_MODEL)) { kpc.setThreadingModel(DEFAULT_THREADING_MODEL); } if (!config.containsKey(THREAD_POOL_SIZE)) { kpc.setThreadPoolSize(DEFAULT_THREAD_POOL_SIZE); } return kpc; } }
KinesisConfigUtil { public static KinesisProducerConfiguration getValidatedProducerConfiguration(Properties config) { checkNotNull(config, "config can not be null"); validateAwsConfiguration(config); if (!config.containsKey(AWSConfigConstants.AWS_REGION)) { throw new IllegalArgumentException(String.format("For FlinkKinesisProducer AWS region ('%s') must be set in the config.", AWSConfigConstants.AWS_REGION)); } KinesisProducerConfiguration kpc = KinesisProducerConfiguration.fromProperties(config); kpc.setRegion(config.getProperty(AWSConfigConstants.AWS_REGION)); kpc.setCredentialsProvider(AWSUtil.getCredentialsProvider(config)); kpc.setCredentialsRefreshDelay(100); if (!config.containsKey(RATE_LIMIT)) { kpc.setRateLimit(DEFAULT_RATE_LIMIT); } if (!config.containsKey(THREADING_MODEL)) { kpc.setThreadingModel(DEFAULT_THREADING_MODEL); } if (!config.containsKey(THREAD_POOL_SIZE)) { kpc.setThreadPoolSize(DEFAULT_THREAD_POOL_SIZE); } return kpc; }  }
KinesisConfigUtil { public static KinesisProducerConfiguration getValidatedProducerConfiguration(Properties config) { checkNotNull(config, "config can not be null"); validateAwsConfiguration(config); if (!config.containsKey(AWSConfigConstants.AWS_REGION)) { throw new IllegalArgumentException(String.format("For FlinkKinesisProducer AWS region ('%s') must be set in the config.", AWSConfigConstants.AWS_REGION)); } KinesisProducerConfiguration kpc = KinesisProducerConfiguration.fromProperties(config); kpc.setRegion(config.getProperty(AWSConfigConstants.AWS_REGION)); kpc.setCredentialsProvider(AWSUtil.getCredentialsProvider(config)); kpc.setCredentialsRefreshDelay(100); if (!config.containsKey(RATE_LIMIT)) { kpc.setRateLimit(DEFAULT_RATE_LIMIT); } if (!config.containsKey(THREADING_MODEL)) { kpc.setThreadingModel(DEFAULT_THREADING_MODEL); } if (!config.containsKey(THREAD_POOL_SIZE)) { kpc.setThreadPoolSize(DEFAULT_THREAD_POOL_SIZE); } return kpc; }  static void validateConsumerConfiguration(Properties config); static Properties replaceDeprecatedProducerKeys(Properties configProps); static Properties backfillConsumerKeys(Properties configProps); static KinesisProducerConfiguration getValidatedProducerConfiguration(Properties config); static void validateAwsConfiguration(Properties config); }
KinesisConfigUtil { public static KinesisProducerConfiguration getValidatedProducerConfiguration(Properties config) { checkNotNull(config, "config can not be null"); validateAwsConfiguration(config); if (!config.containsKey(AWSConfigConstants.AWS_REGION)) { throw new IllegalArgumentException(String.format("For FlinkKinesisProducer AWS region ('%s') must be set in the config.", AWSConfigConstants.AWS_REGION)); } KinesisProducerConfiguration kpc = KinesisProducerConfiguration.fromProperties(config); kpc.setRegion(config.getProperty(AWSConfigConstants.AWS_REGION)); kpc.setCredentialsProvider(AWSUtil.getCredentialsProvider(config)); kpc.setCredentialsRefreshDelay(100); if (!config.containsKey(RATE_LIMIT)) { kpc.setRateLimit(DEFAULT_RATE_LIMIT); } if (!config.containsKey(THREADING_MODEL)) { kpc.setThreadingModel(DEFAULT_THREADING_MODEL); } if (!config.containsKey(THREAD_POOL_SIZE)) { kpc.setThreadPoolSize(DEFAULT_THREAD_POOL_SIZE); } return kpc; }  static void validateConsumerConfiguration(Properties config); static Properties replaceDeprecatedProducerKeys(Properties configProps); static Properties backfillConsumerKeys(Properties configProps); static KinesisProducerConfiguration getValidatedProducerConfiguration(Properties config); static void validateAwsConfiguration(Properties config);  }