Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Allow InitialPositionInStreamExtended to be specified in properties file #804

Merged
merged 3 commits into from
Jun 1, 2021
Merged

Allow InitialPositionInStreamExtended to be specified in properties file #804

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6e0cbb9
Allow InitialPositionInStreamExtended to be specified in properties file
kevioke Apr 2, 2021
362e086
Create shared timestamp for testWithInitialPositionInStreamExtended
kevioke May 10, 2021
afee84d
Add more unit tests for exception cases
kevioke May 10, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions .../src/main/java/software/amazon/kinesis/multilang/config/MultiLangDaemonConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import java.net.URI;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
Expand All @@ -27,6 +28,7 @@
import java.util.function.Function;

import org.apache.commons.beanutils.BeanUtilsBean;
import org.apache.commons.beanutils.ConvertUtils;
import org.apache.commons.beanutils.ConvertUtilsBean;
import org.apache.commons.beanutils.Converter;
import org.apache.commons.beanutils.converters.ArrayConverter;
Expand Down Expand Up @@ -197,6 +199,14 @@ public MultiLangDaemonConfiguration(BeanUtilsBean utilsBean, ConvertUtilsBean co
this.utilsBean = utilsBean;
this.convertUtilsBean = convertUtilsBean;

convertUtilsBean.register(new Converter() {
@Override
public <T> T convert(Class<T> type, Object value) {
Date date = new Date(Long.parseLong(value.toString()) * 1000L);
return type.cast(InitialPositionInStreamExtended.newInitialPositionAtTimestamp(date));
}
}, InitialPositionInStreamExtended.class);

convertUtilsBean.register(new Converter() {
@Override
public <T> T convert(Class<T> type, Object value) {
Expand Down
11 changes: 11 additions & 0 deletions .../test/java/software/amazon/kinesis/multilang/config/KinesisClientLibConfiguratorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import java.io.InputStream;
import java.net.URI;
import java.util.Arrays;
import java.util.Date;
import java.util.HashSet;
import java.util.Optional;
import java.util.Set;
Expand Down Expand Up @@ -93,6 +94,16 @@ public void testWithLongVariables() {
assertEquals(config.getShardSyncIntervalMillis(), 500);
}

@Test
public void testWithInitialPositionInStreamExtended() {
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks like similar tests are in KCL 1.0, but not in 2.0. In addition to this one, can you please add the following test case (testKCLConfigurationWithInvalidInitialPositionInStream) here? You will have to make "config.getInitialPositionInStreamExtended()" public. But that's fine.

https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfigurationTest.java#L319

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice find! Definitely good to test the exception cases as well.

I've updated the tests. I tried to copy the essence of the unit tests you mentioned over, but didn't exactly copy everything word for word since the 1.0 tests seem to be testing the KinesisClientLibConfiguration instead of the MultiLangDaemonConfiguration.

So to recap, I believe the unit tests in this test file should now cover these scenarios:

  1. When InitialPositionInStreamExtended is configured with a timestamp, it sets the AT_TIMESTAMP as well as time in the config.getInitialPositionInStreamExtended(). config.getInitialPositionInStreamExtended() seems to already be public or at least package private for MultiLangDaemonConfiguration. I don't explicitly see the declaration but I'm assuming it's generated via lomboc

    amazon-kinesis-client/amazon-kinesis-client-multilang/src/main/java/software/amazon/kinesis/multilang/config/MultiLangDaemonConfiguration.java

    Line 66 in afee84d

    @Getter
    .
  2. Exceptions should be thrown when InitialPositionInStreamExtended is not a long value
  3. Exceptions should be thrown when InitialPositionInStream is set to AT_TIMESTAMP
  4. TRIM_HORIZON is a valid value for initialPositionInStream
  5. LATEST is a valid value for initialPositionInStream

MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "applicationName = app",
"streamName = 123", "AWSCredentialsProvider = " + credentialName1 + ", " + credentialName2,
"initialPositionInStreamExtended = 1617406032" }, '\n'));

assertEquals(config.getInitialPositionInStreamExtended().getTimestamp(), new Date(1617406032000L));
kevioke marked this conversation as resolved.
Show resolved Hide resolved
assertEquals(config.getInitialPositionInStream(), InitialPositionInStream.AT_TIMESTAMP);
}

@Test
public void testWithUnsupportedClientConfigurationVariables() {
MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(
Expand Down