KinesisConfig extends MapConfig { public KinesisClientLibConfiguration getKinesisClientLibConfig(String system, String stream, String appName) { ClientConfiguration clientConfig = getAWSClientConfig(system); String workerId = appName + "-" + UUID.randomUUID(); InitialPositionInStream startPos = InitialPositionInStream.LATEST; AWSCredentialsProvider provider = credentialsProviderForStream(system, stream); KinesisClientLibConfiguration kinesisClientLibConfiguration = new KinesisClientLibConfiguration(appName, stream, provider, workerId) .withRegionName(getRegion(system, stream).getName()) .withKinesisClientConfig(clientConfig) .withCloudWatchClientConfig(clientConfig) .withDynamoDBClientConfig(clientConfig) .withInitialPositionInStream(startPos) .withCallProcessRecordsEvenForEmptyRecordList(true); setKinesisClientLibConfigs( subset(String.format(CONFIG_SYSTEM_KINESIS_CLIENT_LIB_CONFIG, system)), kinesisClientLibConfiguration); setKinesisClientLibConfigs(subset(String.format(CONFIG_STREAM_KINESIS_CLIENT_LIB_CONFIG, system, stream)), kinesisClientLibConfiguration); return kinesisClientLibConfiguration; } KinesisConfig(Config config); Set<String> getKinesisStreams(String system); KinesisClientLibConfiguration getKinesisClientLibConfig(String system, String stream, String appName); static final String CONFIG_SYSTEM_REGION; static final String CONFIG_STREAM_REGION; static final String CONFIG_STREAM_ACCESS_KEY; static final String CONFIG_STREAM_SECRET_KEY; static final String CONFIG_AWS_CLIENT_CONFIG; static final String CONFIG_PROXY_HOST; static final String DEFAULT_CONFIG_PROXY_HOST; static final String CONFIG_PROXY_PORT; static final int DEFAULT_CONFIG_PROXY_PORT; static final String CONFIG_SYSTEM_KINESIS_CLIENT_LIB_CONFIG; static final String CONFIG_STREAM_KINESIS_CLIENT_LIB_CONFIG; }
@Test public void testKclConfigs() { Map<String, String> kv = new HashMap<>(); String system = "kinesis"; String stream = "kinesis-stream"; String systemConfigPrefix = String.format("systems.%s.", system); kv.put(systemConfigPrefix + "aws.region", "us-east-1"); kv.put(systemConfigPrefix + "aws.kcl.TableName", "sample-table"); kv.put(systemConfigPrefix + "aws.kcl.MaxRecords", "100"); kv.put(systemConfigPrefix + "aws.kcl.CallProcessRecordsEvenForEmptyRecordList", "true"); kv.put(systemConfigPrefix + "aws.kcl.InitialPositionInStream", "TRIM_HORIZON"); kv.put(systemConfigPrefix + "streams.kinesis-stream1.aws.kcl.InitialPositionInStream", "LATEST"); Config config = new MapConfig(kv); KinesisConfig kConfig = new KinesisConfig(config); KinesisClientLibConfiguration kclConfig = kConfig.getKinesisClientLibConfig(system, stream, "sample-app"); assertEquals("sample-table", kclConfig.getTableName()); assertEquals(100, kclConfig.getMaxRecords()); assertTrue(kclConfig.shouldCallProcessRecordsEvenForEmptyRecordList()); assertEquals(InitialPositionInStream.TRIM_HORIZON, kclConfig.getInitialPositionInStream()); kclConfig = kConfig.getKinesisClientLibConfig(system, "kinesis-stream1", "sample-app"); assertEquals(InitialPositionInStream.LATEST, kclConfig.getInitialPositionInStream()); }
@Test public void testgetKCLConfigWithUnknownConfigs() { Map<String, String> kv = new HashMap<>(); kv.put("systems.kinesis.aws.region", "us-east-1"); kv.put("systems.kinesis.streams.kinesis-stream.aws.kcl.random", "value"); Config config = new MapConfig(kv); KinesisConfig kConfig = new KinesisConfig(config); kConfig.getKinesisClientLibConfig("kinesis", "kinesis-stream", "sample-app"); }