@Test public void testRetrievalModeSetForPollingString() throws Exception { MultiLangDaemonConfiguration configuration = baseConfiguration(); utilsBean.setProperty(configuration, "retrievalMode", RetrievalMode.POLLING.name().toLowerCase()); MultiLangDaemonConfiguration.ResolvedConfiguration resolvedConfiguration = configuration .resolvedConfiguration(shardRecordProcessorFactory); assertThat(resolvedConfiguration.getRetrievalConfig().retrievalSpecificConfig(), instanceOf(PollingConfig.class)); }
ResolvedConfiguration resolvedConfiguration(ShardRecordProcessorFactory shardRecordProcessorFactory) { AwsCredentialsProvider kinesisCreds = resolveCredentials(kinesisCredentialsProvider); AwsCredentialsProvider dynamoDbCreds = resolveCredentials(dynamoDBCredentialsProvider); AwsCredentialsProvider cloudwatchCreds = resolveCredentials(cloudWatchCredentialsProvider); updateCredentials(kinesisClient, kinesisCreds, kinesisCreds); updateCredentials(dynamoDbClient, dynamoDbCreds, kinesisCreds); updateCredentials(cloudWatchClient, cloudwatchCreds, kinesisCreds); KinesisAsyncClient kinesisAsyncClient = kinesisClient.build(KinesisAsyncClient.class, this::adjustKinesisHttpConfiguration); DynamoDbAsyncClient dynamoDbAsyncClient = dynamoDbClient.build(DynamoDbAsyncClient.class); CloudWatchAsyncClient cloudWatchAsyncClient = cloudWatchClient.build(CloudWatchAsyncClient.class); ConfigsBuilder configsBuilder = new ConfigsBuilder(streamName, applicationName, kinesisAsyncClient, dynamoDbAsyncClient, cloudWatchAsyncClient, workerIdentifier, shardRecordProcessorFactory); Map<Class<?>, Object> configObjects = new HashMap<>(); addConfigObjects(configObjects, configsBuilder); resolveFields(configObjects, Collections.singleton(ConfigsBuilder.class), Collections.singleton(PollingConfig.class)); CoordinatorConfig coordinatorConfig = configsBuilder.coordinatorConfig(); CheckpointConfig checkpointConfig = configsBuilder.checkpointConfig(); LeaseManagementConfig leaseManagementConfig = configsBuilder.leaseManagementConfig(); LifecycleConfig lifecycleConfig = configsBuilder.lifecycleConfig(); MetricsConfig metricsConfig = configsBuilder.metricsConfig(); ProcessorConfig processorConfig = configsBuilder.processorConfig(); RetrievalConfig retrievalConfig = configsBuilder.retrievalConfig(); addConfigObjects(configObjects, coordinatorConfig, checkpointConfig, leaseManagementConfig, lifecycleConfig, metricsConfig, processorConfig, retrievalConfig); handleRetrievalConfig(retrievalConfig, configsBuilder); resolveFields(configObjects, null, new HashSet<>(Arrays.asList(ConfigsBuilder.class, PollingConfig.class))); return new ResolvedConfiguration(coordinatorConfig, checkpointConfig, leaseManagementConfig, lifecycleConfig, metricsConfig, processorConfig, retrievalConfig); }
MultiLangDaemonConfiguration { ResolvedConfiguration resolvedConfiguration(ShardRecordProcessorFactory shardRecordProcessorFactory) { AwsCredentialsProvider kinesisCreds = resolveCredentials(kinesisCredentialsProvider); AwsCredentialsProvider dynamoDbCreds = resolveCredentials(dynamoDBCredentialsProvider); AwsCredentialsProvider cloudwatchCreds = resolveCredentials(cloudWatchCredentialsProvider); updateCredentials(kinesisClient, kinesisCreds, kinesisCreds); updateCredentials(dynamoDbClient, dynamoDbCreds, kinesisCreds); updateCredentials(cloudWatchClient, cloudwatchCreds, kinesisCreds); KinesisAsyncClient kinesisAsyncClient = kinesisClient.build(KinesisAsyncClient.class, this::adjustKinesisHttpConfiguration); DynamoDbAsyncClient dynamoDbAsyncClient = dynamoDbClient.build(DynamoDbAsyncClient.class); CloudWatchAsyncClient cloudWatchAsyncClient = cloudWatchClient.build(CloudWatchAsyncClient.class); ConfigsBuilder configsBuilder = new ConfigsBuilder(streamName, applicationName, kinesisAsyncClient, dynamoDbAsyncClient, cloudWatchAsyncClient, workerIdentifier, shardRecordProcessorFactory); Map<Class<?>, Object> configObjects = new HashMap<>(); addConfigObjects(configObjects, configsBuilder); resolveFields(configObjects, Collections.singleton(ConfigsBuilder.class), Collections.singleton(PollingConfig.class)); CoordinatorConfig coordinatorConfig = configsBuilder.coordinatorConfig(); CheckpointConfig checkpointConfig = configsBuilder.checkpointConfig(); LeaseManagementConfig leaseManagementConfig = configsBuilder.leaseManagementConfig(); LifecycleConfig lifecycleConfig = configsBuilder.lifecycleConfig(); MetricsConfig metricsConfig = configsBuilder.metricsConfig(); ProcessorConfig processorConfig = configsBuilder.processorConfig(); RetrievalConfig retrievalConfig = configsBuilder.retrievalConfig(); addConfigObjects(configObjects, coordinatorConfig, checkpointConfig, leaseManagementConfig, lifecycleConfig, metricsConfig, processorConfig, retrievalConfig); handleRetrievalConfig(retrievalConfig, configsBuilder); resolveFields(configObjects, null, new HashSet<>(Arrays.asList(ConfigsBuilder.class, PollingConfig.class))); return new ResolvedConfiguration(coordinatorConfig, checkpointConfig, leaseManagementConfig, lifecycleConfig, metricsConfig, processorConfig, retrievalConfig); } }
MultiLangDaemonConfiguration { ResolvedConfiguration resolvedConfiguration(ShardRecordProcessorFactory shardRecordProcessorFactory) { AwsCredentialsProvider kinesisCreds = resolveCredentials(kinesisCredentialsProvider); AwsCredentialsProvider dynamoDbCreds = resolveCredentials(dynamoDBCredentialsProvider); AwsCredentialsProvider cloudwatchCreds = resolveCredentials(cloudWatchCredentialsProvider); updateCredentials(kinesisClient, kinesisCreds, kinesisCreds); updateCredentials(dynamoDbClient, dynamoDbCreds, kinesisCreds); updateCredentials(cloudWatchClient, cloudwatchCreds, kinesisCreds); KinesisAsyncClient kinesisAsyncClient = kinesisClient.build(KinesisAsyncClient.class, this::adjustKinesisHttpConfiguration); DynamoDbAsyncClient dynamoDbAsyncClient = dynamoDbClient.build(DynamoDbAsyncClient.class); CloudWatchAsyncClient cloudWatchAsyncClient = cloudWatchClient.build(CloudWatchAsyncClient.class); ConfigsBuilder configsBuilder = new ConfigsBuilder(streamName, applicationName, kinesisAsyncClient, dynamoDbAsyncClient, cloudWatchAsyncClient, workerIdentifier, shardRecordProcessorFactory); Map<Class<?>, Object> configObjects = new HashMap<>(); addConfigObjects(configObjects, configsBuilder); resolveFields(configObjects, Collections.singleton(ConfigsBuilder.class), Collections.singleton(PollingConfig.class)); CoordinatorConfig coordinatorConfig = configsBuilder.coordinatorConfig(); CheckpointConfig checkpointConfig = configsBuilder.checkpointConfig(); LeaseManagementConfig leaseManagementConfig = configsBuilder.leaseManagementConfig(); LifecycleConfig lifecycleConfig = configsBuilder.lifecycleConfig(); MetricsConfig metricsConfig = configsBuilder.metricsConfig(); ProcessorConfig processorConfig = configsBuilder.processorConfig(); RetrievalConfig retrievalConfig = configsBuilder.retrievalConfig(); addConfigObjects(configObjects, coordinatorConfig, checkpointConfig, leaseManagementConfig, lifecycleConfig, metricsConfig, processorConfig, retrievalConfig); handleRetrievalConfig(retrievalConfig, configsBuilder); resolveFields(configObjects, null, new HashSet<>(Arrays.asList(ConfigsBuilder.class, PollingConfig.class))); return new ResolvedConfiguration(coordinatorConfig, checkpointConfig, leaseManagementConfig, lifecycleConfig, metricsConfig, processorConfig, retrievalConfig); } MultiLangDaemonConfiguration(BeanUtilsBean utilsBean, ConvertUtilsBean convertUtilsBean); }
MultiLangDaemonConfiguration { ResolvedConfiguration resolvedConfiguration(ShardRecordProcessorFactory shardRecordProcessorFactory) { AwsCredentialsProvider kinesisCreds = resolveCredentials(kinesisCredentialsProvider); AwsCredentialsProvider dynamoDbCreds = resolveCredentials(dynamoDBCredentialsProvider); AwsCredentialsProvider cloudwatchCreds = resolveCredentials(cloudWatchCredentialsProvider); updateCredentials(kinesisClient, kinesisCreds, kinesisCreds); updateCredentials(dynamoDbClient, dynamoDbCreds, kinesisCreds); updateCredentials(cloudWatchClient, cloudwatchCreds, kinesisCreds); KinesisAsyncClient kinesisAsyncClient = kinesisClient.build(KinesisAsyncClient.class, this::adjustKinesisHttpConfiguration); DynamoDbAsyncClient dynamoDbAsyncClient = dynamoDbClient.build(DynamoDbAsyncClient.class); CloudWatchAsyncClient cloudWatchAsyncClient = cloudWatchClient.build(CloudWatchAsyncClient.class); ConfigsBuilder configsBuilder = new ConfigsBuilder(streamName, applicationName, kinesisAsyncClient, dynamoDbAsyncClient, cloudWatchAsyncClient, workerIdentifier, shardRecordProcessorFactory); Map<Class<?>, Object> configObjects = new HashMap<>(); addConfigObjects(configObjects, configsBuilder); resolveFields(configObjects, Collections.singleton(ConfigsBuilder.class), Collections.singleton(PollingConfig.class)); CoordinatorConfig coordinatorConfig = configsBuilder.coordinatorConfig(); CheckpointConfig checkpointConfig = configsBuilder.checkpointConfig(); LeaseManagementConfig leaseManagementConfig = configsBuilder.leaseManagementConfig(); LifecycleConfig lifecycleConfig = configsBuilder.lifecycleConfig(); MetricsConfig metricsConfig = configsBuilder.metricsConfig(); ProcessorConfig processorConfig = configsBuilder.processorConfig(); RetrievalConfig retrievalConfig = configsBuilder.retrievalConfig(); addConfigObjects(configObjects, coordinatorConfig, checkpointConfig, leaseManagementConfig, lifecycleConfig, metricsConfig, processorConfig, retrievalConfig); handleRetrievalConfig(retrievalConfig, configsBuilder); resolveFields(configObjects, null, new HashSet<>(Arrays.asList(ConfigsBuilder.class, PollingConfig.class))); return new ResolvedConfiguration(coordinatorConfig, checkpointConfig, leaseManagementConfig, lifecycleConfig, metricsConfig, processorConfig, retrievalConfig); } MultiLangDaemonConfiguration(BeanUtilsBean utilsBean, ConvertUtilsBean convertUtilsBean); void setWorkerId(String workerId); InitialPositionInStream getInitialPositionInStream(); void setInitialPositionInStream(InitialPositionInStream initialPositionInStream); String[] getMetricsEnabledDimensions(); void setMetricsEnabledDimensions(String[] dimensions); void setAWSCredentialsProvider(String providerString); void setAWSCredentialsProviderDynamoDB(String providerString); void setAWSCredentialsProviderCloudWatch(String providerString); void setRegionName(Region region); void setKinesisEndpoint(String endpoint); void setDynamoDBEndpoint(String endpoint); Scheduler build(ShardRecordProcessorFactory shardRecordProcessorFactory); }
MultiLangDaemonConfiguration { ResolvedConfiguration resolvedConfiguration(ShardRecordProcessorFactory shardRecordProcessorFactory) { AwsCredentialsProvider kinesisCreds = resolveCredentials(kinesisCredentialsProvider); AwsCredentialsProvider dynamoDbCreds = resolveCredentials(dynamoDBCredentialsProvider); AwsCredentialsProvider cloudwatchCreds = resolveCredentials(cloudWatchCredentialsProvider); updateCredentials(kinesisClient, kinesisCreds, kinesisCreds); updateCredentials(dynamoDbClient, dynamoDbCreds, kinesisCreds); updateCredentials(cloudWatchClient, cloudwatchCreds, kinesisCreds); KinesisAsyncClient kinesisAsyncClient = kinesisClient.build(KinesisAsyncClient.class, this::adjustKinesisHttpConfiguration); DynamoDbAsyncClient dynamoDbAsyncClient = dynamoDbClient.build(DynamoDbAsyncClient.class); CloudWatchAsyncClient cloudWatchAsyncClient = cloudWatchClient.build(CloudWatchAsyncClient.class); ConfigsBuilder configsBuilder = new ConfigsBuilder(streamName, applicationName, kinesisAsyncClient, dynamoDbAsyncClient, cloudWatchAsyncClient, workerIdentifier, shardRecordProcessorFactory); Map<Class<?>, Object> configObjects = new HashMap<>(); addConfigObjects(configObjects, configsBuilder); resolveFields(configObjects, Collections.singleton(ConfigsBuilder.class), Collections.singleton(PollingConfig.class)); CoordinatorConfig coordinatorConfig = configsBuilder.coordinatorConfig(); CheckpointConfig checkpointConfig = configsBuilder.checkpointConfig(); LeaseManagementConfig leaseManagementConfig = configsBuilder.leaseManagementConfig(); LifecycleConfig lifecycleConfig = configsBuilder.lifecycleConfig(); MetricsConfig metricsConfig = configsBuilder.metricsConfig(); ProcessorConfig processorConfig = configsBuilder.processorConfig(); RetrievalConfig retrievalConfig = configsBuilder.retrievalConfig(); addConfigObjects(configObjects, coordinatorConfig, checkpointConfig, leaseManagementConfig, lifecycleConfig, metricsConfig, processorConfig, retrievalConfig); handleRetrievalConfig(retrievalConfig, configsBuilder); resolveFields(configObjects, null, new HashSet<>(Arrays.asList(ConfigsBuilder.class, PollingConfig.class))); return new ResolvedConfiguration(coordinatorConfig, checkpointConfig, leaseManagementConfig, lifecycleConfig, metricsConfig, processorConfig, retrievalConfig); } MultiLangDaemonConfiguration(BeanUtilsBean utilsBean, ConvertUtilsBean convertUtilsBean); void setWorkerId(String workerId); InitialPositionInStream getInitialPositionInStream(); void setInitialPositionInStream(InitialPositionInStream initialPositionInStream); String[] getMetricsEnabledDimensions(); void setMetricsEnabledDimensions(String[] dimensions); void setAWSCredentialsProvider(String providerString); void setAWSCredentialsProviderDynamoDB(String providerString); void setAWSCredentialsProviderCloudWatch(String providerString); void setRegionName(Region region); void setKinesisEndpoint(String endpoint); void setDynamoDBEndpoint(String endpoint); Scheduler build(ShardRecordProcessorFactory shardRecordProcessorFactory);  }