You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2017/09/26 16:04:43 UTC
[1/9] flink git commit: [FLINK-4047] [docs] Fix documentation about
determinism of KeySelectors
Repository: flink
Updated Branches:
refs/heads/master 2b82578ab -> 58320e8a6
[FLINK-4047] [docs] Fix documentation about determinism of KeySelectors
This closes #4659.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/58320e8a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/58320e8a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/58320e8a
Branch: refs/heads/master
Commit: 58320e8a6118651573bb062a1082ed5671c3a11e
Parents: b054841
Author: zhangminglei <zm...@163.com>
Authored: Fri Sep 15 15:06:44 2017 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Tue Sep 26 18:04:07 2017 +0200
----------------------------------------------------------------------
docs/dev/api_concepts.md | 2 +-
.../org/apache/flink/api/java/functions/KeySelector.java | 9 +++++----
2 files changed, 6 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/58320e8a/docs/dev/api_concepts.md
----------------------------------------------------------------------
diff --git a/docs/dev/api_concepts.md b/docs/dev/api_concepts.md
index c447f27..eaf0e17 100644
--- a/docs/dev/api_concepts.md
+++ b/docs/dev/api_concepts.md
@@ -450,7 +450,7 @@ These are valid field expressions for the example code above:
{:.no_toc}
An additional way to define keys are "key selector" functions. A key selector function
-takes a single element as input and returns the key for the element. The key can be of any type and be derived from arbitrary computations.
+takes a single element as input and returns the key for the element. The key can be of any type and be derived from deterministic computations.
The following example shows a key selector function that simply returns the field of an object:
http://git-wip-us.apache.org/repos/asf/flink/blob/58320e8a/flink-core/src/main/java/org/apache/flink/api/java/functions/KeySelector.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/functions/KeySelector.java b/flink-core/src/main/java/org/apache/flink/api/java/functions/KeySelector.java
index da3b3e2..63e76a9 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/functions/KeySelector.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/functions/KeySelector.java
@@ -24,10 +24,11 @@ import org.apache.flink.api.common.functions.Function;
import java.io.Serializable;
/**
- * The {@link KeySelector} allows to use arbitrary objects for operations such as
- * reduce, reduceGroup, join, coGoup, etc.
+ * The {@link KeySelector} allows to use deterministic objects for operations such as
+ * reduce, reduceGroup, join, coGoup, etc. If invoked multiple times on the same object,
+ * the returned key must be the same.
*
- * The extractor takes an object and returns the key for that object.
+ * The extractor takes an object and returns the deterministic key for that object.
*
* @param <IN> Type of objects to extract the key from.
* @param <KEY> Type of key.
@@ -36,7 +37,7 @@ import java.io.Serializable;
public interface KeySelector<IN, KEY> extends Function, Serializable {
/**
- * User-defined function that extracts the key from an arbitrary object.
+ * User-defined function that deterministically extracts the key from an object.
*
* For example for a class:
* <pre>
[6/9] flink git commit: [FLINK-7393] [kinesis] Move unit tests of
KinesisConfigUtil from FlinkKinesisConsumerTest to KinesisConfigUtilTest
Posted by tz...@apache.org.
[FLINK-7393] [kinesis] Move unit tests of KinesisConfigUtil from FlinkKinesisConsumerTest to KinesisConfigUtilTest
This closes #4708.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ff417e76
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ff417e76
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ff417e76
Branch: refs/heads/master
Commit: ff417e762032b100a74a6e9ee215618f4ac9db3e
Parents: 415e7d0
Author: Bowen Li <bo...@gmail.com>
Authored: Fri Sep 22 16:10:56 2017 -0700
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Tue Sep 26 18:04:07 2017 +0200
----------------------------------------------------------------------
.../kinesis/FlinkKinesisConsumerTest.java | 417 ------------------
.../kinesis/util/KinesisConfigUtilTest.java | 421 +++++++++++++++++++
2 files changed, 421 insertions(+), 417 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ff417e76/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
index 4a007d5..78ca160 100644
--- a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
+++ b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
@@ -32,7 +32,6 @@ import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContextSynchronousImpl;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kinesis.config.AWSConfigConstants;
-import org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants;
import org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher;
import org.apache.flink.streaming.connectors.kinesis.model.KinesisStreamShard;
import org.apache.flink.streaming.connectors.kinesis.model.KinesisStreamShardState;
@@ -73,7 +72,6 @@ import java.util.UUID;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
@@ -90,421 +88,6 @@ public class FlinkKinesisConsumerTest {
private ExpectedException exception = ExpectedException.none();
// ----------------------------------------------------------------------
- // FlinkKinesisConsumer.validateAwsConfiguration() tests
- // ----------------------------------------------------------------------
-
- @Test
- public void testMissingAwsRegionInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("The AWS region ('" + AWSConfigConstants.AWS_REGION + "') must be set in the config.");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(AWSConfigConstants.AWS_ACCESS_KEY_ID, "accessKey");
- testConfig.setProperty(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
-
- KinesisConfigUtil.validateAwsConfiguration(testConfig);
- }
-
- @Test
- public void testUnrecognizableAwsRegionInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid AWS region");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(AWSConfigConstants.AWS_REGION, "wrongRegionId");
- testConfig.setProperty(AWSConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
-
- KinesisConfigUtil.validateAwsConfiguration(testConfig);
- }
-
- @Test
- public void testCredentialProviderTypeSetToBasicButNoCredentialSetInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Please set values for AWS Access Key ID ('" + AWSConfigConstants.AWS_ACCESS_KEY_ID + "') " +
- "and Secret Key ('" + AWSConfigConstants.AWS_SECRET_ACCESS_KEY + "') when using the BASIC AWS credential provider type.");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(AWSConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(AWSConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
-
- KinesisConfigUtil.validateAwsConfiguration(testConfig);
- }
-
- @Test
- public void testUnrecognizableCredentialProviderTypeInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid AWS Credential Provider Type");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(AWSConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(AWSConfigConstants.AWS_CREDENTIALS_PROVIDER, "wrongProviderType");
- testConfig.setProperty(AWSConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
-
- KinesisConfigUtil.validateAwsConfiguration(testConfig);
- }
-
- // ----------------------------------------------------------------------
- // FlinkKinesisConsumer.validateConsumerConfiguration() tests
- // ----------------------------------------------------------------------
-
- @Test
- public void testUnrecognizableStreamInitPositionTypeInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid initial position in stream");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "wrongInitPosition");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testStreamInitPositionTypeSetToAtTimestampButNoInitTimestampSetInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Please set value for initial timestamp ('"
- + ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP + "') when using AT_TIMESTAMP initial position.");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableDateForInitialTimestampInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for initial timestamp for AT_TIMESTAMP initial position in stream.");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, "unparsableDate");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testIllegalValueForInitialTimestampInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for initial timestamp for AT_TIMESTAMP initial position in stream.");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, "-1.0");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testDateStringForValidateOptionDateProperty() {
- String timestamp = "2016-04-04T19:58:46.480-00:00";
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, timestamp);
-
- try {
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- } catch (Exception e) {
- e.printStackTrace();
- fail();
- }
- }
-
- @Test
- public void testUnixTimestampForValidateOptionDateProperty() {
- String unixTimestamp = "1459799926.480";
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, unixTimestamp);
-
- try {
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- } catch (Exception e) {
- e.printStackTrace();
- fail();
- }
- }
-
- @Test
- public void testInvalidPatternForInitialTimestampInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for initial timestamp for AT_TIMESTAMP initial position in stream.");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, "2016-03-14");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_TIMESTAMP_DATE_FORMAT, "InvalidPattern");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableDateForUserDefinedDateFormatForInitialTimestampInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for initial timestamp for AT_TIMESTAMP initial position in stream.");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, "stillUnparsable");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_TIMESTAMP_DATE_FORMAT, "yyyy-MM-dd");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testDateStringForUserDefinedDateFormatForValidateOptionDateProperty() {
- String unixTimestamp = "2016-04-04";
- String pattern = "yyyy-MM-dd";
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, unixTimestamp);
- testConfig.setProperty(ConsumerConfigConstants.STREAM_TIMESTAMP_DATE_FORMAT, pattern);
-
- try {
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- } catch (Exception e) {
- e.printStackTrace();
- fail();
- }
- }
-
- @Test
- public void testUnparsableLongForDescribeStreamBackoffBaseMillisInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for describe stream operation base backoff milliseconds");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_BASE, "unparsableLong");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableLongForDescribeStreamBackoffMaxMillisInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for describe stream operation max backoff milliseconds");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_MAX, "unparsableLong");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableDoubleForDescribeStreamBackoffExponentialConstantInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for describe stream operation backoff exponential constant");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_EXPONENTIAL_CONSTANT, "unparsableDouble");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableIntForGetRecordsRetriesInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for maximum retry attempts for getRecords shard operation");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_RETRIES, "unparsableInt");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableIntForGetRecordsMaxCountInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for maximum records per getRecords shard operation");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_MAX, "unparsableInt");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableLongForGetRecordsBackoffBaseMillisInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for get records operation base backoff milliseconds");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_BASE, "unparsableLong");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableLongForGetRecordsBackoffMaxMillisInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for get records operation max backoff milliseconds");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_MAX, "unparsableLong");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableDoubleForGetRecordsBackoffExponentialConstantInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for get records operation backoff exponential constant");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_EXPONENTIAL_CONSTANT, "unparsableDouble");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableLongForGetRecordsIntervalMillisInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for getRecords sleep interval in milliseconds");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_INTERVAL_MILLIS, "unparsableLong");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableIntForGetShardIteratorRetriesInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for maximum retry attempts for getShardIterator shard operation");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.SHARD_GETITERATOR_RETRIES, "unparsableInt");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableLongForGetShardIteratorBackoffBaseMillisInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for get shard iterator operation base backoff milliseconds");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_BASE, "unparsableLong");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableLongForGetShardIteratorBackoffMaxMillisInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for get shard iterator operation max backoff milliseconds");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_MAX, "unparsableLong");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableDoubleForGetShardIteratorBackoffExponentialConstantInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for get shard iterator operation backoff exponential constant");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_EXPONENTIAL_CONSTANT, "unparsableDouble");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- @Test
- public void testUnparsableLongForShardDiscoveryIntervalMillisInConfig() {
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Invalid value given for shard discovery sleep interval in milliseconds");
-
- Properties testConfig = new Properties();
- testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
- testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
- testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- testConfig.setProperty(ConsumerConfigConstants.SHARD_DISCOVERY_INTERVAL_MILLIS, "unparsableLong");
-
- KinesisConfigUtil.validateConsumerConfiguration(testConfig);
- }
-
- // ----------------------------------------------------------------------
// Tests related to state initialization
// ----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ff417e76/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtilTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtilTest.java b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtilTest.java
index a576748..ddb300e 100644
--- a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtilTest.java
+++ b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtilTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.connectors.kinesis.util;
import org.apache.flink.streaming.connectors.kinesis.config.AWSConfigConstants;
+import org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants;
import org.apache.flink.streaming.connectors.kinesis.config.ProducerConfigConstants;
import com.amazonaws.services.kinesis.producer.KinesisProducerConfiguration;
@@ -32,6 +33,7 @@ import org.powermock.modules.junit4.PowerMockRunner;
import java.util.Properties;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
/**
* Tests for KinesisConfigUtil.
@@ -42,6 +44,10 @@ public class KinesisConfigUtilTest {
@Rule
private ExpectedException exception = ExpectedException.none();
+ // ----------------------------------------------------------------------
+ // getValidatedProducerConfiguration() tests
+ // ----------------------------------------------------------------------
+
@Test
public void testUnparsableLongForProducerConfiguration() {
exception.expect(IllegalArgumentException.class);
@@ -125,4 +131,419 @@ public class KinesisConfigUtilTest {
assertEquals("1", replacedConfig.getProperty(KinesisConfigUtil.AGGREGATION_MAX_COUNT));
assertEquals("2", replacedConfig.getProperty(KinesisConfigUtil.COLLECTION_MAX_COUNT));
}
+
+ // ----------------------------------------------------------------------
+ // validateAwsConfiguration() tests
+ // ----------------------------------------------------------------------
+
+ @Test
+ public void testMissingAwsRegionInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("The AWS region ('" + AWSConfigConstants.AWS_REGION + "') must be set in the config.");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(AWSConfigConstants.AWS_ACCESS_KEY_ID, "accessKey");
+ testConfig.setProperty(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+
+ KinesisConfigUtil.validateAwsConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnrecognizableAwsRegionInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid AWS region");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(AWSConfigConstants.AWS_REGION, "wrongRegionId");
+ testConfig.setProperty(AWSConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+
+ KinesisConfigUtil.validateAwsConfiguration(testConfig);
+ }
+
+ @Test
+ public void testCredentialProviderTypeSetToBasicButNoCredentialSetInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Please set values for AWS Access Key ID ('" + AWSConfigConstants.AWS_ACCESS_KEY_ID + "') " +
+ "and Secret Key ('" + AWSConfigConstants.AWS_SECRET_ACCESS_KEY + "') when using the BASIC AWS credential provider type.");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(AWSConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(AWSConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
+
+ KinesisConfigUtil.validateAwsConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnrecognizableCredentialProviderTypeInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid AWS Credential Provider Type");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(AWSConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(AWSConfigConstants.AWS_CREDENTIALS_PROVIDER, "wrongProviderType");
+ testConfig.setProperty(AWSConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+
+ KinesisConfigUtil.validateAwsConfiguration(testConfig);
+ }
+
+ // ----------------------------------------------------------------------
+ // validateConsumerConfiguration() tests
+ // ----------------------------------------------------------------------
+
+ @Test
+ public void testUnrecognizableStreamInitPositionTypeInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid initial position in stream");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "wrongInitPosition");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testStreamInitPositionTypeSetToAtTimestampButNoInitTimestampSetInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Please set value for initial timestamp ('"
+ + ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP + "') when using AT_TIMESTAMP initial position.");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableDateForInitialTimestampInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for initial timestamp for AT_TIMESTAMP initial position in stream.");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, "unparsableDate");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testIllegalValueForInitialTimestampInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for initial timestamp for AT_TIMESTAMP initial position in stream.");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, "-1.0");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testDateStringForValidateOptionDateProperty() {
+ String timestamp = "2016-04-04T19:58:46.480-00:00";
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, timestamp);
+
+ try {
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail();
+ }
+ }
+
+ @Test
+ public void testUnixTimestampForValidateOptionDateProperty() {
+ String unixTimestamp = "1459799926.480";
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, unixTimestamp);
+
+ try {
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail();
+ }
+ }
+
+ @Test
+ public void testInvalidPatternForInitialTimestampInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for initial timestamp for AT_TIMESTAMP initial position in stream.");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, "2016-03-14");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_TIMESTAMP_DATE_FORMAT, "InvalidPattern");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableDateForUserDefinedDateFormatForInitialTimestampInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for initial timestamp for AT_TIMESTAMP initial position in stream.");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, "stillUnparsable");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_TIMESTAMP_DATE_FORMAT, "yyyy-MM-dd");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testDateStringForUserDefinedDateFormatForValidateOptionDateProperty() {
+ String unixTimestamp = "2016-04-04";
+ String pattern = "yyyy-MM-dd";
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, unixTimestamp);
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_TIMESTAMP_DATE_FORMAT, pattern);
+
+ try {
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail();
+ }
+ }
+
+ @Test
+ public void testUnparsableLongForDescribeStreamBackoffBaseMillisInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for describe stream operation base backoff milliseconds");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_BASE, "unparsableLong");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableLongForDescribeStreamBackoffMaxMillisInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for describe stream operation max backoff milliseconds");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_MAX, "unparsableLong");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableDoubleForDescribeStreamBackoffExponentialConstantInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for describe stream operation backoff exponential constant");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_EXPONENTIAL_CONSTANT, "unparsableDouble");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableIntForGetRecordsRetriesInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for maximum retry attempts for getRecords shard operation");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_RETRIES, "unparsableInt");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableIntForGetRecordsMaxCountInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for maximum records per getRecords shard operation");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_MAX, "unparsableInt");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableLongForGetRecordsBackoffBaseMillisInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for get records operation base backoff milliseconds");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_BASE, "unparsableLong");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableLongForGetRecordsBackoffMaxMillisInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for get records operation max backoff milliseconds");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_MAX, "unparsableLong");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableDoubleForGetRecordsBackoffExponentialConstantInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for get records operation backoff exponential constant");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_EXPONENTIAL_CONSTANT, "unparsableDouble");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableLongForGetRecordsIntervalMillisInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for getRecords sleep interval in milliseconds");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_INTERVAL_MILLIS, "unparsableLong");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableIntForGetShardIteratorRetriesInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for maximum retry attempts for getShardIterator shard operation");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.SHARD_GETITERATOR_RETRIES, "unparsableInt");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableLongForGetShardIteratorBackoffBaseMillisInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for get shard iterator operation base backoff milliseconds");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_BASE, "unparsableLong");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableLongForGetShardIteratorBackoffMaxMillisInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for get shard iterator operation max backoff milliseconds");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_MAX, "unparsableLong");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableDoubleForGetShardIteratorBackoffExponentialConstantInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for get shard iterator operation backoff exponential constant");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_EXPONENTIAL_CONSTANT, "unparsableDouble");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
+ @Test
+ public void testUnparsableLongForShardDiscoveryIntervalMillisInConfig() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Invalid value given for shard discovery sleep interval in milliseconds");
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId");
+ testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+ testConfig.setProperty(ConsumerConfigConstants.SHARD_DISCOVERY_INTERVAL_MILLIS, "unparsableLong");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
}
[2/9] flink git commit: [FLINK-7654] [rabbitmq] Update RabbitMQ Java
client version to 4.2.0
Posted by tz...@apache.org.
[FLINK-7654] [rabbitmq] Update RabbitMQ Java client version to 4.2.0
This closes #4694.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/bd0841e7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/bd0841e7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/bd0841e7
Branch: refs/heads/master
Commit: bd0841e7b8febb7a78f119db3193e81b96fcd84b
Parents: 4ba3eec
Author: yew1eb <ye...@gmail.com>
Authored: Thu Sep 21 14:38:52 2017 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Tue Sep 26 18:04:07 2017 +0200
----------------------------------------------------------------------
docs/dev/connectors/rabbitmq.md | 6 +-
.../flink-connector-rabbitmq/README.md | 6 +-
.../flink-connector-rabbitmq/pom.xml | 2 +-
.../streaming/connectors/rabbitmq/RMQSink.java | 13 +-
.../connectors/rabbitmq/RMQSource.java | 4 +-
.../connectors/rabbitmq/RMQSinkTest.java | 133 +++++++++++++++++++
.../connectors/rabbitmq/RMQSourceTest.java | 3 +-
.../connectors/rabbitmq/common/RMQSinkTest.java | 133 -------------------
8 files changed, 152 insertions(+), 148 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/bd0841e7/docs/dev/connectors/rabbitmq.md
----------------------------------------------------------------------
diff --git a/docs/dev/connectors/rabbitmq.md b/docs/dev/connectors/rabbitmq.md
index d3360e4..5780892 100644
--- a/docs/dev/connectors/rabbitmq.md
+++ b/docs/dev/connectors/rabbitmq.md
@@ -26,16 +26,14 @@ under the License.
# License of the RabbitMQ Connector
Flink's RabbitMQ connector defines a Maven dependency on the
-"RabbitMQ AMQP Java Client", licensed under the
-[Mozilla Public License v1.1 (MPL 1.1)](https://www.mozilla.org/en-US/MPL/1.1/).
+"RabbitMQ AMQP Java Client", is triple-licensed under the Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2 ("GPL") and the Apache License version 2 ("ASL").
Flink itself neither reuses source code from the "RabbitMQ AMQP Java Client"
nor packages binaries from the "RabbitMQ AMQP Java Client".
Users that create and publish derivative work based on Flink's
RabbitMQ connector (thereby re-distributing the "RabbitMQ AMQP Java Client")
-must be aware that this may be subject to conditions declared
-in the Mozilla Public License v1.1 (MPL 1.1).
+must be aware that this may be subject to conditions declared in the Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2 ("GPL") and the Apache License version 2 ("ASL").
# RabbitMQ Connector
http://git-wip-us.apache.org/repos/asf/flink/blob/bd0841e7/flink-connectors/flink-connector-rabbitmq/README.md
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-rabbitmq/README.md b/flink-connectors/flink-connector-rabbitmq/README.md
index 4a0e783..de8d1d8 100644
--- a/flink-connectors/flink-connector-rabbitmq/README.md
+++ b/flink-connectors/flink-connector-rabbitmq/README.md
@@ -1,13 +1,11 @@
# License of the Rabbit MQ Connector
Flink's RabbitMQ connector defines a Maven dependency on the
-"RabbitMQ AMQP Java Client", licensed under the
-Mozilla Public License v1.1 (MPL 1.1).
+"RabbitMQ AMQP Java Client", is triple-licensed under the Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2 ("GPL") and the Apache License version 2 ("ASL").
Flink itself neither reuses source code from the "RabbitMQ AMQP Java Client"
nor packages binaries from the "RabbitMQ AMQP Java Client".
Users that create and publish derivative work based on Flink's
RabbitMQ connector (thereby re-distributing the "RabbitMQ AMQP Java Client")
-must be aware that this may be subject to conditions declared
-in the Mozilla Public License v1.1 (MPL 1.1).
+must be aware that this may be subject to conditions declared in the Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2 ("GPL") and the Apache License version 2 ("ASL").
http://git-wip-us.apache.org/repos/asf/flink/blob/bd0841e7/flink-connectors/flink-connector-rabbitmq/pom.xml
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-rabbitmq/pom.xml b/flink-connectors/flink-connector-rabbitmq/pom.xml
index 38337f4..2bb3404 100644
--- a/flink-connectors/flink-connector-rabbitmq/pom.xml
+++ b/flink-connectors/flink-connector-rabbitmq/pom.xml
@@ -37,7 +37,7 @@ under the License.
<!-- Allow users to pass custom connector versions -->
<properties>
- <rabbitmq.version>3.3.1</rabbitmq.version>
+ <rabbitmq.version>4.2.0</rabbitmq.version>
</properties>
<dependencies>
http://git-wip-us.apache.org/repos/asf/flink/blob/bd0841e7/flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java b/flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
index 48675c5..b3c6f8c 100644
--- a/flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
+++ b/flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
@@ -29,6 +29,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.concurrent.TimeoutException;
/**
* A Sink for publishing data into RabbitMQ.
@@ -117,15 +118,19 @@ public class RMQSink<IN> extends RichSinkFunction<IN> {
@Override
public void close() {
- IOException t = null;
+ Exception t = null;
try {
- channel.close();
- } catch (IOException e) {
+ if (channel != null) {
+ channel.close();
+ }
+ } catch (IOException | TimeoutException e) {
t = e;
}
try {
- connection.close();
+ if (connection != null) {
+ connection.close();
+ }
} catch (IOException e) {
if (t != null) {
LOG.warn("Both channel and connection closing failed. Logging channel exception and failing with connection exception", t);
http://git-wip-us.apache.org/repos/asf/flink/blob/bd0841e7/flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java b/flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
index 12a35f1..5018bcf 100644
--- a/flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
+++ b/flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
@@ -178,7 +178,9 @@ public class RMQSource<OUT> extends MultipleIdsMessageAcknowledgingSourceBase<OU
public void close() throws Exception {
super.close();
try {
- connection.close();
+ if (connection != null) {
+ connection.close();
+ }
} catch (IOException e) {
throw new RuntimeException("Error while closing RMQ connection with " + queueName
+ " at " + rmqConnectionConfig.getHost(), e);
http://git-wip-us.apache.org/repos/asf/flink/blob/bd0841e7/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java b/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
new file mode 100644
index 0000000..93f884b
--- /dev/null
+++ b/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.rabbitmq;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.sink.SinkContextUtil;
+import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
+import org.apache.flink.streaming.util.serialization.SerializationSchema;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for the {@link RMQSink}.
+ */
+public class RMQSinkTest {
+
+ private static final String QUEUE_NAME = "queue";
+ private static final String MESSAGE_STR = "msg";
+ private static final byte[] MESSAGE = new byte[1];
+
+ private RMQConnectionConfig rmqConnectionConfig;
+ private ConnectionFactory connectionFactory;
+ private Connection connection;
+ private Channel channel;
+ private SerializationSchema<String> serializationSchema;
+
+ @Before
+ public void before() throws Exception {
+ serializationSchema = spy(new DummySerializationSchema());
+ rmqConnectionConfig = mock(RMQConnectionConfig.class);
+ connectionFactory = mock(ConnectionFactory.class);
+ connection = mock(Connection.class);
+ channel = mock(Channel.class);
+
+ when(rmqConnectionConfig.getConnectionFactory()).thenReturn(connectionFactory);
+ when(connectionFactory.newConnection()).thenReturn(connection);
+ when(connection.createChannel()).thenReturn(channel);
+ }
+
+ @Test
+ public void openCallDeclaresQueue() throws Exception {
+ createRMQSink();
+
+ verify(channel).queueDeclare(QUEUE_NAME, false, false, false, null);
+ }
+
+ @Test
+ public void throwExceptionIfChannelIsNull() throws Exception {
+ when(connection.createChannel()).thenReturn(null);
+ try {
+ createRMQSink();
+ } catch (RuntimeException ex) {
+ assertEquals("None of RabbitMQ channels are available", ex.getMessage());
+ }
+ }
+
+ private RMQSink<String> createRMQSink() throws Exception {
+ RMQSink rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ rmqSink.open(new Configuration());
+ return rmqSink;
+ }
+
+ @Test
+ public void invokePublishBytesToQueue() throws Exception {
+ RMQSink<String> rmqSink = createRMQSink();
+
+ rmqSink.invoke(MESSAGE_STR, SinkContextUtil.forTimestamp(0));
+ verify(serializationSchema).serialize(MESSAGE_STR);
+ verify(channel).basicPublish("", QUEUE_NAME, null, MESSAGE);
+ }
+
+ @Test(expected = RuntimeException.class)
+ public void exceptionDuringPublishingIsNotIgnored() throws Exception {
+ RMQSink<String> rmqSink = createRMQSink();
+
+ doThrow(IOException.class).when(channel).basicPublish("", QUEUE_NAME, null, MESSAGE);
+ rmqSink.invoke("msg", SinkContextUtil.forTimestamp(0));
+ }
+
+ @Test
+ public void exceptionDuringPublishingIsIgnoredIfLogFailuresOnly() throws Exception {
+ RMQSink<String> rmqSink = createRMQSink();
+ rmqSink.setLogFailuresOnly(true);
+
+ doThrow(IOException.class).when(channel).basicPublish("", QUEUE_NAME, null, MESSAGE);
+ rmqSink.invoke("msg", SinkContextUtil.forTimestamp(0));
+ }
+
+ @Test
+ public void closeAllResources() throws Exception {
+ RMQSink<String> rmqSink = createRMQSink();
+
+ rmqSink.close();
+
+ verify(channel).close();
+ verify(connection).close();
+ }
+
+ private class DummySerializationSchema implements SerializationSchema<String> {
+ @Override
+ public byte[] serialize(String element) {
+ return MESSAGE;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/bd0841e7/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java b/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java
index f180e78..0996355 100644
--- a/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java
+++ b/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java
@@ -53,6 +53,7 @@ import java.io.IOException;
import java.util.ArrayDeque;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.TimeoutException;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
@@ -404,7 +405,7 @@ public class RMQSourceTest {
try {
Mockito.when(connectionFactory.newConnection()).thenReturn(connection);
Mockito.when(connection.createChannel()).thenReturn(Mockito.mock(Channel.class));
- } catch (IOException e) {
+ } catch (IOException | TimeoutException e) {
fail("Test environment couldn't be created.");
}
return connectionFactory;
http://git-wip-us.apache.org/repos/asf/flink/blob/bd0841e7/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQSinkTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQSinkTest.java b/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQSinkTest.java
deleted file mode 100644
index 4fb6097..0000000
--- a/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQSinkTest.java
+++ /dev/null
@@ -1,133 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.connectors.rabbitmq.common;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.functions.sink.SinkContextUtil;
-import org.apache.flink.streaming.connectors.rabbitmq.RMQSink;
-import org.apache.flink.streaming.util.serialization.SerializationSchema;
-
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.io.IOException;
-
-import static org.junit.Assert.assertEquals;
-import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-/**
- * Tests for the {@link RMQSink}.
- */
-public class RMQSinkTest {
-
- private static final String QUEUE_NAME = "queue";
- private static final String MESSAGE_STR = "msg";
- private static final byte[] MESSAGE = new byte[1];
-
- private RMQConnectionConfig rmqConnectionConfig;
- private ConnectionFactory connectionFactory;
- private Connection connection;
- private Channel channel;
- private SerializationSchema<String> serializationSchema;
-
- @Before
- public void before() throws Exception {
- serializationSchema = spy(new DummySerializationSchema());
- rmqConnectionConfig = mock(RMQConnectionConfig.class);
- connectionFactory = mock(ConnectionFactory.class);
- connection = mock(Connection.class);
- channel = mock(Channel.class);
-
- when(rmqConnectionConfig.getConnectionFactory()).thenReturn(connectionFactory);
- when(connectionFactory.newConnection()).thenReturn(connection);
- when(connection.createChannel()).thenReturn(channel);
- }
-
- @Test
- public void openCallDeclaresQueue() throws Exception {
- createRMQSink();
-
- verify(channel).queueDeclare(QUEUE_NAME, false, false, false, null);
- }
-
- @Test
- public void throwExceptionIfChannelIsNull() throws Exception {
- when(connection.createChannel()).thenReturn(null);
- try {
- createRMQSink();
- } catch (RuntimeException ex) {
- assertEquals("None of RabbitMQ channels are available", ex.getMessage());
- }
- }
-
- private RMQSink<String> createRMQSink() throws Exception {
- RMQSink rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
- rmqSink.open(new Configuration());
- return rmqSink;
- }
-
- @Test
- public void invokePublishBytesToQueue() throws Exception {
- RMQSink<String> rmqSink = createRMQSink();
-
- rmqSink.invoke(MESSAGE_STR, SinkContextUtil.forTimestamp(0));
- verify(serializationSchema).serialize(MESSAGE_STR);
- verify(channel).basicPublish("", QUEUE_NAME, null, MESSAGE);
- }
-
- @Test(expected = RuntimeException.class)
- public void exceptionDuringPublishingIsNotIgnored() throws Exception {
- RMQSink<String> rmqSink = createRMQSink();
-
- doThrow(IOException.class).when(channel).basicPublish("", QUEUE_NAME, null, MESSAGE);
- rmqSink.invoke("msg", SinkContextUtil.forTimestamp(0));
- }
-
- @Test
- public void exceptionDuringPublishingIsIgnoredIfLogFailuresOnly() throws Exception {
- RMQSink<String> rmqSink = createRMQSink();
- rmqSink.setLogFailuresOnly(true);
-
- doThrow(IOException.class).when(channel).basicPublish("", QUEUE_NAME, null, MESSAGE);
- rmqSink.invoke("msg", SinkContextUtil.forTimestamp(0));
- }
-
- @Test
- public void closeAllResources() throws Exception {
- RMQSink<String> rmqSink = createRMQSink();
-
- rmqSink.close();
-
- verify(channel).close();
- verify(connection).close();
- }
-
- private class DummySerializationSchema implements SerializationSchema<String> {
- @Override
- public byte[] serialize(String element) {
- return MESSAGE;
- }
- }
-}
[5/9] flink git commit: [FLINK-7647] [flip6] Port
JobManagerConfigHandler to new REST endpoint
Posted by tz...@apache.org.
[FLINK-7647] [flip6] Port JobManagerConfigHandler to new REST endpoint
This commit lets the JobManagerConfigHandler implement the
LegacyRestHandler interface in order to be ported to the new REST
endpoint. This includes the introduction of ClusterConfiguration
response body and ClusterConfigurationHeaders.
The DispatcherRestEndpoint now also registers the
JobManagerConfigHandler.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/97ff043f
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/97ff043f
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/97ff043f
Branch: refs/heads/master
Commit: 97ff043fda06147bfd6441495d893fce7baf882e
Parents: 6c97230
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Wed Sep 20 18:25:28 2017 +0200
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Tue Sep 26 18:04:07 2017 +0200
----------------------------------------------------------------------
.../dispatcher/DispatcherRestEndpoint.java | 21 ++++-
.../entrypoint/SessionClusterEntrypoint.java | 1 +
.../handler/legacy/JobManagerConfigHandler.java | 97 ++++++++++++--------
.../messages/ClusterConfigurationInfo.java | 59 ++++++++++++
.../messages/ClusterConfigurationInfoEntry.java | 81 ++++++++++++++++
.../ClusterConfigurationInfoHeaders.java | 71 ++++++++++++++
.../legacy/JobManagerConfigHandlerTest.java | 3 +-
.../messages/ClusterConfigurationInfoTest.java | 53 +++++++++++
8 files changed, 345 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/97ff043f/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
index dff5df8..3e8a4cb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.dispatcher;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
import org.apache.flink.runtime.rest.RestServerEndpoint;
import org.apache.flink.runtime.rest.RestServerEndpointConfiguration;
@@ -29,10 +30,13 @@ import org.apache.flink.runtime.rest.handler.RestHandlerSpecification;
import org.apache.flink.runtime.rest.handler.legacy.ClusterOverviewHandler;
import org.apache.flink.runtime.rest.handler.legacy.CurrentJobsOverviewHandler;
import org.apache.flink.runtime.rest.handler.legacy.DashboardConfigHandler;
+import org.apache.flink.runtime.rest.handler.legacy.JobManagerConfigHandler;
import org.apache.flink.runtime.rest.handler.legacy.files.StaticFileServerHandler;
import org.apache.flink.runtime.rest.handler.legacy.files.WebContentHandlerSpecification;
+import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfo;
import org.apache.flink.runtime.rest.handler.legacy.messages.DashboardConfiguration;
import org.apache.flink.runtime.rest.handler.legacy.messages.StatusOverviewWithVersion;
+import org.apache.flink.runtime.rest.messages.ClusterConfigurationInfoHeaders;
import org.apache.flink.runtime.rest.messages.ClusterOverviewHeaders;
import org.apache.flink.runtime.rest.messages.CurrentJobsOverviewHandlerHeaders;
import org.apache.flink.runtime.rest.messages.DashboardConfigurationHeaders;
@@ -58,16 +62,19 @@ import java.util.concurrent.Executor;
public class DispatcherRestEndpoint extends RestServerEndpoint {
private final GatewayRetriever<DispatcherGateway> leaderRetriever;
+ private final Configuration clusterConfiguration;
private final RestHandlerConfiguration restConfiguration;
private final Executor executor;
public DispatcherRestEndpoint(
- RestServerEndpointConfiguration configuration,
+ RestServerEndpointConfiguration endpointConfiguration,
GatewayRetriever<DispatcherGateway> leaderRetriever,
+ Configuration clusterConfiguration,
RestHandlerConfiguration restConfiguration,
Executor executor) {
- super(configuration);
+ super(endpointConfiguration);
this.leaderRetriever = Preconditions.checkNotNull(leaderRetriever);
+ this.clusterConfiguration = Preconditions.checkNotNull(clusterConfiguration);
this.restConfiguration = Preconditions.checkNotNull(restConfiguration);
this.executor = Preconditions.checkNotNull(executor);
}
@@ -107,6 +114,15 @@ public class DispatcherRestEndpoint extends RestServerEndpoint {
true,
true));
+ LegacyRestHandlerAdapter<DispatcherGateway, ClusterConfigurationInfo, EmptyMessageParameters> clusterConfigurationHandler = new LegacyRestHandlerAdapter<>(
+ restAddressFuture,
+ leaderRetriever,
+ timeout,
+ ClusterConfigurationInfoHeaders.getInstance(),
+ new JobManagerConfigHandler(
+ executor,
+ clusterConfiguration));
+
final File tmpDir = restConfiguration.getTmpDir();
Optional<StaticFileServerHandler<DispatcherGateway>> optWebContent;
@@ -123,6 +139,7 @@ public class DispatcherRestEndpoint extends RestServerEndpoint {
}
handlers.add(Tuple2.of(ClusterOverviewHeaders.getInstance(), clusterOverviewHandler));
+ handlers.add(Tuple2.of(ClusterConfigurationInfoHeaders.getInstance(), clusterConfigurationHandler));
handlers.add(Tuple2.of(DashboardConfigurationHeaders.getInstance(), dashboardConfigurationHandler));
handlers.add(Tuple2.of(CurrentJobsOverviewHandlerHeaders.getInstance(), currentJobsOverviewHandler));
http://git-wip-us.apache.org/repos/asf/flink/blob/97ff043f/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
index c4da6db..be34b0c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
@@ -159,6 +159,7 @@ public abstract class SessionClusterEntrypoint extends ClusterEntrypoint {
return new DispatcherRestEndpoint(
RestServerEndpointConfiguration.fromConfiguration(configuration),
dispatcherGatewayRetriever,
+ configuration,
RestHandlerConfiguration.fromConfiguration(configuration),
executor);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/97ff043f/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandler.java
index d638467..6c0c158 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandler.java
@@ -19,8 +19,17 @@
package org.apache.flink.runtime.rest.handler.legacy;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.dispatcher.DispatcherGateway;
import org.apache.flink.runtime.jobmaster.JobManagerGateway;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.LegacyRestHandler;
+import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfo;
+import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfoEntry;
+import org.apache.flink.runtime.rest.messages.ClusterConfigurationInfoHeaders;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.Preconditions;
import com.fasterxml.jackson.core.JsonGenerator;
@@ -34,55 +43,67 @@ import java.util.concurrent.Executor;
/**
* Returns the Job Manager's configuration.
*/
-public class JobManagerConfigHandler extends AbstractJsonRequestHandler {
+public class JobManagerConfigHandler extends AbstractJsonRequestHandler
+ implements LegacyRestHandler<DispatcherGateway, ClusterConfigurationInfo, EmptyMessageParameters> {
- private static final String JOBMANAGER_CONFIG_REST_PATH = "/jobmanager/config";
-
- private final Configuration config;
+ private final ClusterConfigurationInfo clusterConfig;
+ private final String clusterConfigJson;
public JobManagerConfigHandler(Executor executor, Configuration config) {
super(executor);
- this.config = config;
+
+ Preconditions.checkNotNull(config);
+ this.clusterConfig = ClusterConfigurationInfo.from(config);
+ this.clusterConfigJson = createConfigJson(config);
}
@Override
public String[] getPaths() {
- return new String[]{JOBMANAGER_CONFIG_REST_PATH};
+ return new String[]{ClusterConfigurationInfoHeaders.CLUSTER_CONFIG_REST_PATH};
}
@Override
- public CompletableFuture<String> handleJsonRequest(Map<String, String> pathParams, Map<String, String> queryParams, JobManagerGateway jobManagerGateway) {
- return CompletableFuture.supplyAsync(
- () -> {
- try {
- StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.JACKSON_FACTORY.createGenerator(writer);
-
- gen.writeStartArray();
- for (String key : config.keySet()) {
- gen.writeStartObject();
- gen.writeStringField("key", key);
-
- // Mask key values which contain sensitive information
- if (key.toLowerCase().contains("password")) {
- String value = config.getString(key, null);
- if (value != null) {
- value = "******";
- }
- gen.writeStringField("value", value);
- } else {
- gen.writeStringField("value", config.getString(key, null));
- }
- gen.writeEndObject();
- }
- gen.writeEndArray();
-
- gen.close();
- return writer.toString();
- } catch (IOException e) {
- throw new CompletionException(new FlinkException("Could not write configuration.", e));
+ public CompletableFuture<ClusterConfigurationInfo> handleRequest(
+ HandlerRequest<EmptyRequestBody, EmptyMessageParameters> request,
+ DispatcherGateway gateway) {
+
+ return CompletableFuture.completedFuture(clusterConfig);
+ }
+
+ @Override
+ public CompletableFuture<String> handleJsonRequest(
+ Map<String, String> pathParams,
+ Map<String, String> queryParams,
+ JobManagerGateway jobManagerGateway) {
+
+ return CompletableFuture.completedFuture(clusterConfigJson);
+ }
+
+ private static String createConfigJson(Configuration config) {
+ try {
+ StringWriter writer = new StringWriter();
+ JsonGenerator gen = JsonFactory.JACKSON_FACTORY.createGenerator(writer);
+
+ gen.writeStartArray();
+ for (String key : config.keySet()) {
+ gen.writeStartObject();
+ gen.writeStringField(ClusterConfigurationInfoEntry.FIELD_NAME_CONFIG_KEY, key);
+
+ String value = config.getString(key, null);
+ // Mask key values which contain sensitive information
+ if (value != null && key.toLowerCase().contains("password")) {
+ value = "******";
}
- },
- executor);
+ gen.writeStringField(ClusterConfigurationInfoEntry.FIELD_NAME_CONFIG_VALUE, value);
+
+ gen.writeEndObject();
+ }
+ gen.writeEndArray();
+
+ gen.close();
+ return writer.toString();
+ } catch (IOException e) {
+ throw new CompletionException(new FlinkException("Could not write configuration.", e));
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/97ff043f/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfo.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfo.java
new file mode 100644
index 0000000..8b87e63
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfo.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.legacy.messages;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.rest.handler.legacy.JobManagerConfigHandler;
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+
+import java.util.ArrayList;
+
+/**
+ * Response of the {@link JobManagerConfigHandler}, respresented as a list
+ * of key-value pairs of the cluster {@link Configuration}.
+ */
+public class ClusterConfigurationInfo extends ArrayList<ClusterConfigurationInfoEntry> implements ResponseBody {
+
+ private static final long serialVersionUID = -1170348873871206964L;
+
+ // a default constructor is required for collection type marshalling
+ public ClusterConfigurationInfo() {}
+
+ public ClusterConfigurationInfo(int initialEntries) {
+ super(initialEntries);
+ }
+
+ public static ClusterConfigurationInfo from(Configuration config) {
+ ClusterConfigurationInfo clusterConfig = new ClusterConfigurationInfo(config.keySet().size());
+
+ for (String key : config.keySet()) {
+ String value = config.getString(key, null);
+
+ // Mask key values which contain sensitive information
+ if (value != null && key.toLowerCase().contains("password")) {
+ value = "******";
+ }
+
+ clusterConfig.add(new ClusterConfigurationInfoEntry(key, value));
+ }
+
+ return clusterConfig;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/97ff043f/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoEntry.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoEntry.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoEntry.java
new file mode 100644
index 0000000..19e3821
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoEntry.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.legacy.messages;
+
+import org.apache.flink.util.Preconditions;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Objects;
+
+/**
+ * A single key-value pair entry in the {@link ClusterConfigurationInfo} response.
+ */
+public class ClusterConfigurationInfoEntry {
+
+ public static final String FIELD_NAME_CONFIG_KEY = "key";
+ public static final String FIELD_NAME_CONFIG_VALUE = "value";
+
+ @JsonProperty(FIELD_NAME_CONFIG_KEY)
+ private final String key;
+
+ @JsonProperty(FIELD_NAME_CONFIG_VALUE)
+ private final String value;
+
+ @JsonCreator
+ public ClusterConfigurationInfoEntry(
+ @JsonProperty(FIELD_NAME_CONFIG_KEY) String key,
+ @JsonProperty(FIELD_NAME_CONFIG_VALUE) String value) {
+ this.key = Preconditions.checkNotNull(key);
+ this.value = Preconditions.checkNotNull(value);
+ }
+
+ public String getKey() {
+ return key;
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ ClusterConfigurationInfoEntry that = (ClusterConfigurationInfoEntry) o;
+ return key.equals(that.key) && value.equals(that.value);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(key, value);
+ }
+
+ @Override
+ public String toString() {
+ return "(" + key + "," + value + ")";
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/97ff043f/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java
new file mode 100644
index 0000000..764e497
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.handler.legacy.JobManagerConfigHandler;
+import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfo;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+/**
+ * Message headers for the {@link JobManagerConfigHandler}.
+ */
+public final class ClusterConfigurationInfoHeaders implements MessageHeaders<EmptyRequestBody, ClusterConfigurationInfo, EmptyMessageParameters> {
+
+ private static final ClusterConfigurationInfoHeaders INSTANCE = new ClusterConfigurationInfoHeaders();
+
+ public static final String CLUSTER_CONFIG_REST_PATH = "/jobmanager/config";
+
+ private ClusterConfigurationInfoHeaders() {}
+
+ @Override
+ public Class<EmptyRequestBody> getRequestClass() {
+ return EmptyRequestBody.class;
+ }
+
+ @Override
+ public HttpMethodWrapper getHttpMethod() {
+ return HttpMethodWrapper.GET;
+ }
+
+ @Override
+ public String getTargetRestEndpointURL() {
+ return CLUSTER_CONFIG_REST_PATH;
+ }
+
+ @Override
+ public Class<ClusterConfigurationInfo> getResponseClass() {
+ return ClusterConfigurationInfo.class;
+ }
+
+ @Override
+ public HttpResponseStatus getResponseStatusCode() {
+ return HttpResponseStatus.OK;
+ }
+
+ @Override
+ public EmptyMessageParameters getUnresolvedMessageParameters() {
+ return EmptyMessageParameters.getInstance();
+ }
+
+ public static ClusterConfigurationInfoHeaders getInstance() {
+ return INSTANCE;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/97ff043f/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandlerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandlerTest.java
index 03ddb73..f21b1ed 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandlerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandlerTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.rest.handler.legacy;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.concurrent.Executors;
import org.junit.Assert;
@@ -29,7 +30,7 @@ import org.junit.Test;
public class JobManagerConfigHandlerTest {
@Test
public void testGetPaths() {
- JobManagerConfigHandler handler = new JobManagerConfigHandler(Executors.directExecutor(), null);
+ JobManagerConfigHandler handler = new JobManagerConfigHandler(Executors.directExecutor(), new Configuration());
String[] paths = handler.getPaths();
Assert.assertEquals(1, paths.length);
Assert.assertEquals("/jobmanager/config", paths[0]);
http://git-wip-us.apache.org/repos/asf/flink/blob/97ff043f/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoTest.java
new file mode 100644
index 0000000..1d7d9a4
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoTest.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.legacy.messages;
+
+import org.apache.flink.runtime.rest.util.RestMapperUtils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link ClusterConfigurationInfo}.
+ */
+public class ClusterConfigurationInfoTest {
+
+ /**
+ * Tests that we can marshal and unmarshal {@link ClusterConfigurationInfo} objects.
+ */
+ @Test
+ public void testJsonMarshalling() throws JsonProcessingException {
+ final ClusterConfigurationInfo expected = new ClusterConfigurationInfo(2);
+ expected.add(new ClusterConfigurationInfoEntry("key1", "value1"));
+ expected.add(new ClusterConfigurationInfoEntry("key2", "value2"));
+
+ final ObjectMapper objectMapper = RestMapperUtils.getStrictObjectMapper();
+
+ JsonNode marshaled = objectMapper.valueToTree(expected);
+
+ final ClusterConfigurationInfo unmarshaled = objectMapper.treeToValue(marshaled, ClusterConfigurationInfo.class);
+
+ assertEquals(expected, unmarshaled);
+ }
+}
[7/9] flink git commit: [FLINK-7647] [flip6] Rename
JobManagerConfigHandler to ClusterConfigHandler
Posted by tz...@apache.org.
[FLINK-7647] [flip6] Rename JobManagerConfigHandler to ClusterConfigHandler
The original naming wouldn't make sense for the FLIP-6 redesign, since
we would have multiple per-job JobManagers for each cluster, which
shares the same configuration.
The REST path is still left untouched and not part of this commit, as
that would involve more changes in flink-runtime-web.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/2b0008c5
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/2b0008c5
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/2b0008c5
Branch: refs/heads/master
Commit: 2b0008c50d27640e5a8a8be491528f4c248fdf32
Parents: 97ff043
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Wed Sep 20 18:51:27 2017 +0200
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Tue Sep 26 18:04:07 2017 +0200
----------------------------------------------------------------------
.../runtime/webmonitor/WebRuntimeMonitor.java | 4 +-
.../dispatcher/DispatcherRestEndpoint.java | 4 +-
.../handler/legacy/ClusterConfigHandler.java | 109 +++++++++++++++++++
.../handler/legacy/JobManagerConfigHandler.java | 109 -------------------
.../messages/ClusterConfigurationInfo.java | 4 +-
.../ClusterConfigurationInfoHeaders.java | 6 +-
.../legacy/ClusterConfigHandlerTest.java | 38 +++++++
.../legacy/JobManagerConfigHandlerTest.java | 38 -------
8 files changed, 157 insertions(+), 155 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/2b0008c5/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
index cd128de..1af6ab6 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
@@ -28,6 +28,7 @@ import org.apache.flink.runtime.jobmaster.JobManagerGateway;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.net.SSLUtils;
import org.apache.flink.runtime.rest.handler.WebHandler;
+import org.apache.flink.runtime.rest.handler.legacy.ClusterConfigHandler;
import org.apache.flink.runtime.rest.handler.legacy.ClusterOverviewHandler;
import org.apache.flink.runtime.rest.handler.legacy.ConstantTextHandler;
import org.apache.flink.runtime.rest.handler.legacy.CurrentJobIdsHandler;
@@ -40,7 +41,6 @@ import org.apache.flink.runtime.rest.handler.legacy.JobCancellationWithSavepoint
import org.apache.flink.runtime.rest.handler.legacy.JobConfigHandler;
import org.apache.flink.runtime.rest.handler.legacy.JobDetailsHandler;
import org.apache.flink.runtime.rest.handler.legacy.JobExceptionsHandler;
-import org.apache.flink.runtime.rest.handler.legacy.JobManagerConfigHandler;
import org.apache.flink.runtime.rest.handler.legacy.JobPlanHandler;
import org.apache.flink.runtime.rest.handler.legacy.JobStoppingHandler;
import org.apache.flink.runtime.rest.handler.legacy.JobVertexAccumulatorsHandler;
@@ -244,7 +244,7 @@ public class WebRuntimeMonitor implements WebMonitor {
get(router, new ClusterOverviewHandler(executor, DEFAULT_REQUEST_TIMEOUT));
// job manager configuration
- get(router, new JobManagerConfigHandler(executor, config));
+ get(router, new ClusterConfigHandler(executor, config));
// overview over jobs
get(router, new CurrentJobsOverviewHandler(executor, DEFAULT_REQUEST_TIMEOUT, true, true));
http://git-wip-us.apache.org/repos/asf/flink/blob/2b0008c5/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
index 3e8a4cb..17f5616 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
@@ -27,10 +27,10 @@ import org.apache.flink.runtime.rest.RestServerEndpointConfiguration;
import org.apache.flink.runtime.rest.handler.LegacyRestHandlerAdapter;
import org.apache.flink.runtime.rest.handler.RestHandlerConfiguration;
import org.apache.flink.runtime.rest.handler.RestHandlerSpecification;
+import org.apache.flink.runtime.rest.handler.legacy.ClusterConfigHandler;
import org.apache.flink.runtime.rest.handler.legacy.ClusterOverviewHandler;
import org.apache.flink.runtime.rest.handler.legacy.CurrentJobsOverviewHandler;
import org.apache.flink.runtime.rest.handler.legacy.DashboardConfigHandler;
-import org.apache.flink.runtime.rest.handler.legacy.JobManagerConfigHandler;
import org.apache.flink.runtime.rest.handler.legacy.files.StaticFileServerHandler;
import org.apache.flink.runtime.rest.handler.legacy.files.WebContentHandlerSpecification;
import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfo;
@@ -119,7 +119,7 @@ public class DispatcherRestEndpoint extends RestServerEndpoint {
leaderRetriever,
timeout,
ClusterConfigurationInfoHeaders.getInstance(),
- new JobManagerConfigHandler(
+ new ClusterConfigHandler(
executor,
clusterConfiguration));
http://git-wip-us.apache.org/repos/asf/flink/blob/2b0008c5/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandler.java
new file mode 100644
index 0000000..d3ea160
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandler.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.legacy;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.dispatcher.DispatcherGateway;
+import org.apache.flink.runtime.jobmaster.JobManagerGateway;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.LegacyRestHandler;
+import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfo;
+import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfoEntry;
+import org.apache.flink.runtime.rest.messages.ClusterConfigurationInfoHeaders;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.Preconditions;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+
+import java.io.IOException;
+import java.io.StringWriter;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.Executor;
+
+/**
+ * Returns the Job Manager's configuration.
+ */
+public class ClusterConfigHandler extends AbstractJsonRequestHandler
+ implements LegacyRestHandler<DispatcherGateway, ClusterConfigurationInfo, EmptyMessageParameters> {
+
+ private final ClusterConfigurationInfo clusterConfig;
+ private final String clusterConfigJson;
+
+ public ClusterConfigHandler(Executor executor, Configuration config) {
+ super(executor);
+
+ Preconditions.checkNotNull(config);
+ this.clusterConfig = ClusterConfigurationInfo.from(config);
+ this.clusterConfigJson = createConfigJson(config);
+ }
+
+ @Override
+ public String[] getPaths() {
+ return new String[]{ClusterConfigurationInfoHeaders.CLUSTER_CONFIG_REST_PATH};
+ }
+
+ @Override
+ public CompletableFuture<ClusterConfigurationInfo> handleRequest(
+ HandlerRequest<EmptyRequestBody, EmptyMessageParameters> request,
+ DispatcherGateway gateway) {
+
+ return CompletableFuture.completedFuture(clusterConfig);
+ }
+
+ @Override
+ public CompletableFuture<String> handleJsonRequest(
+ Map<String, String> pathParams,
+ Map<String, String> queryParams,
+ JobManagerGateway jobManagerGateway) {
+
+ return CompletableFuture.completedFuture(clusterConfigJson);
+ }
+
+ private static String createConfigJson(Configuration config) {
+ try {
+ StringWriter writer = new StringWriter();
+ JsonGenerator gen = JsonFactory.JACKSON_FACTORY.createGenerator(writer);
+
+ gen.writeStartArray();
+ for (String key : config.keySet()) {
+ gen.writeStartObject();
+ gen.writeStringField(ClusterConfigurationInfoEntry.FIELD_NAME_CONFIG_KEY, key);
+
+ String value = config.getString(key, null);
+ // Mask key values which contain sensitive information
+ if (value != null && key.toLowerCase().contains("password")) {
+ value = "******";
+ }
+ gen.writeStringField(ClusterConfigurationInfoEntry.FIELD_NAME_CONFIG_VALUE, value);
+
+ gen.writeEndObject();
+ }
+ gen.writeEndArray();
+
+ gen.close();
+ return writer.toString();
+ } catch (IOException e) {
+ throw new CompletionException(new FlinkException("Could not write configuration.", e));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/2b0008c5/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandler.java
deleted file mode 100644
index 6c0c158..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandler.java
+++ /dev/null
@@ -1,109 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.rest.handler.legacy;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.dispatcher.DispatcherGateway;
-import org.apache.flink.runtime.jobmaster.JobManagerGateway;
-import org.apache.flink.runtime.rest.handler.HandlerRequest;
-import org.apache.flink.runtime.rest.handler.LegacyRestHandler;
-import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfo;
-import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfoEntry;
-import org.apache.flink.runtime.rest.messages.ClusterConfigurationInfoHeaders;
-import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
-import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
-import org.apache.flink.util.FlinkException;
-import org.apache.flink.util.Preconditions;
-
-import com.fasterxml.jackson.core.JsonGenerator;
-
-import java.io.IOException;
-import java.io.StringWriter;
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionException;
-import java.util.concurrent.Executor;
-
-/**
- * Returns the Job Manager's configuration.
- */
-public class JobManagerConfigHandler extends AbstractJsonRequestHandler
- implements LegacyRestHandler<DispatcherGateway, ClusterConfigurationInfo, EmptyMessageParameters> {
-
- private final ClusterConfigurationInfo clusterConfig;
- private final String clusterConfigJson;
-
- public JobManagerConfigHandler(Executor executor, Configuration config) {
- super(executor);
-
- Preconditions.checkNotNull(config);
- this.clusterConfig = ClusterConfigurationInfo.from(config);
- this.clusterConfigJson = createConfigJson(config);
- }
-
- @Override
- public String[] getPaths() {
- return new String[]{ClusterConfigurationInfoHeaders.CLUSTER_CONFIG_REST_PATH};
- }
-
- @Override
- public CompletableFuture<ClusterConfigurationInfo> handleRequest(
- HandlerRequest<EmptyRequestBody, EmptyMessageParameters> request,
- DispatcherGateway gateway) {
-
- return CompletableFuture.completedFuture(clusterConfig);
- }
-
- @Override
- public CompletableFuture<String> handleJsonRequest(
- Map<String, String> pathParams,
- Map<String, String> queryParams,
- JobManagerGateway jobManagerGateway) {
-
- return CompletableFuture.completedFuture(clusterConfigJson);
- }
-
- private static String createConfigJson(Configuration config) {
- try {
- StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.JACKSON_FACTORY.createGenerator(writer);
-
- gen.writeStartArray();
- for (String key : config.keySet()) {
- gen.writeStartObject();
- gen.writeStringField(ClusterConfigurationInfoEntry.FIELD_NAME_CONFIG_KEY, key);
-
- String value = config.getString(key, null);
- // Mask key values which contain sensitive information
- if (value != null && key.toLowerCase().contains("password")) {
- value = "******";
- }
- gen.writeStringField(ClusterConfigurationInfoEntry.FIELD_NAME_CONFIG_VALUE, value);
-
- gen.writeEndObject();
- }
- gen.writeEndArray();
-
- gen.close();
- return writer.toString();
- } catch (IOException e) {
- throw new CompletionException(new FlinkException("Could not write configuration.", e));
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/2b0008c5/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfo.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfo.java
index 8b87e63..bf527d9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfo.java
@@ -19,13 +19,13 @@
package org.apache.flink.runtime.rest.handler.legacy.messages;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.rest.handler.legacy.JobManagerConfigHandler;
+import org.apache.flink.runtime.rest.handler.legacy.ClusterConfigHandler;
import org.apache.flink.runtime.rest.messages.ResponseBody;
import java.util.ArrayList;
/**
- * Response of the {@link JobManagerConfigHandler}, respresented as a list
+ * Response of the {@link ClusterConfigHandler}, respresented as a list
* of key-value pairs of the cluster {@link Configuration}.
*/
public class ClusterConfigurationInfo extends ArrayList<ClusterConfigurationInfoEntry> implements ResponseBody {
http://git-wip-us.apache.org/repos/asf/flink/blob/2b0008c5/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java
index 764e497..0ad56b1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java
@@ -19,18 +19,20 @@
package org.apache.flink.runtime.rest.messages;
import org.apache.flink.runtime.rest.HttpMethodWrapper;
-import org.apache.flink.runtime.rest.handler.legacy.JobManagerConfigHandler;
+import org.apache.flink.runtime.rest.handler.legacy.ClusterConfigHandler;
import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfo;
import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
/**
- * Message headers for the {@link JobManagerConfigHandler}.
+ * Message headers for the {@link ClusterConfigHandler}.
*/
public final class ClusterConfigurationInfoHeaders implements MessageHeaders<EmptyRequestBody, ClusterConfigurationInfo, EmptyMessageParameters> {
private static final ClusterConfigurationInfoHeaders INSTANCE = new ClusterConfigurationInfoHeaders();
+ // TODO this REST path is inappropriately set due to legacy design reasons, and ideally should be '/config';
+ // TODO changing it would require corresponding path changes in flink-runtime-web
public static final String CLUSTER_CONFIG_REST_PATH = "/jobmanager/config";
private ClusterConfigurationInfoHeaders() {}
http://git-wip-us.apache.org/repos/asf/flink/blob/2b0008c5/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandlerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandlerTest.java
new file mode 100644
index 0000000..5094f01
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandlerTest.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.legacy;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.concurrent.Executors;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Tests for the ClusterConfigHandler.
+ */
+public class ClusterConfigHandlerTest {
+ @Test
+ public void testGetPaths() {
+ ClusterConfigHandler handler = new ClusterConfigHandler(Executors.directExecutor(), new Configuration());
+ String[] paths = handler.getPaths();
+ Assert.assertEquals(1, paths.length);
+ Assert.assertEquals("/jobmanager/config", paths[0]);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/2b0008c5/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandlerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandlerTest.java
deleted file mode 100644
index f21b1ed..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandlerTest.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.rest.handler.legacy;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.concurrent.Executors;
-
-import org.junit.Assert;
-import org.junit.Test;
-
-/**
- * Tests for the JobManagerConfigHandler.
- */
-public class JobManagerConfigHandlerTest {
- @Test
- public void testGetPaths() {
- JobManagerConfigHandler handler = new JobManagerConfigHandler(Executors.directExecutor(), new Configuration());
- String[] paths = handler.getPaths();
- Assert.assertEquals(1, paths.length);
- Assert.assertEquals("/jobmanager/config", paths[0]);
- }
-}
[3/9] flink git commit: [hotfix] [doc] Fix incorrect javaStream
access on Scala DataStream API in docs
Posted by tz...@apache.org.
[hotfix] [doc] Fix incorrect javaStream access on Scala DataStream API in docs
This closes #4704.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6c972302
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6c972302
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6c972302
Branch: refs/heads/master
Commit: 6c9723020fd4da7840370d3c9b152ac920320dcf
Parents: ff417e7
Author: Bang, Phiradet <ph...@rakuten.com>
Authored: Fri Sep 22 15:35:37 2017 +0900
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Tue Sep 26 18:04:07 2017 +0200
----------------------------------------------------------------------
docs/dev/datastream_api.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6c972302/docs/dev/datastream_api.md
----------------------------------------------------------------------
diff --git a/docs/dev/datastream_api.md b/docs/dev/datastream_api.md
index 0fc6033..307679d 100644
--- a/docs/dev/datastream_api.md
+++ b/docs/dev/datastream_api.md
@@ -619,7 +619,7 @@ import org.apache.flink.contrib.streaming.DataStreamUtils
import scala.collection.JavaConverters.asScalaIteratorConverter
val myResult: DataStream[(String, Int)] = ...
-val myOutput: Iterator[(String, Int)] = DataStreamUtils.collect(myResult.getJavaStream).asScala
+val myOutput: Iterator[(String, Int)] = DataStreamUtils.collect(myResult.javaStream).asScala
{% endhighlight %}
</div>
</div>
[9/9] flink git commit: [FLINK-7647] [flip6] Introduce test base for
REST response marshalling
Posted by tz...@apache.org.
[FLINK-7647] [flip6] Introduce test base for REST response marshalling
Introduces a common test base that for all REST responses, a subclass
should be implemented to verify that the response can be correctly
marshalled and unmarshalled.
This closes #4691.
This closes #4720.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4ba3eecd
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4ba3eecd
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4ba3eecd
Branch: refs/heads/master
Commit: 4ba3eecd8f1111c39af3f1949827b3d73addec77
Parents: 2b0008c
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Wed Sep 20 19:07:11 2017 +0200
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Tue Sep 26 18:04:07 2017 +0200
----------------------------------------------------------------------
.../messages/ClusterConfigurationInfoTest.java | 32 +++-------
.../messages/DashboardConfigurationTest.java | 34 +++-------
.../RestResponseMarshallingTestBase.java | 65 ++++++++++++++++++++
.../messages/StatusOverviewWithVersionTest.java | 34 +++-------
4 files changed, 92 insertions(+), 73 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4ba3eecd/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoTest.java
index 1d7d9a4..8e7092b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoTest.java
@@ -18,36 +18,22 @@
package org.apache.flink.runtime.rest.handler.legacy.messages;
-import org.apache.flink.runtime.rest.util.RestMapperUtils;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-
/**
* Tests for the {@link ClusterConfigurationInfo}.
*/
-public class ClusterConfigurationInfoTest {
+public class ClusterConfigurationInfoTest extends RestResponseMarshallingTestBase<ClusterConfigurationInfo> {
- /**
- * Tests that we can marshal and unmarshal {@link ClusterConfigurationInfo} objects.
- */
- @Test
- public void testJsonMarshalling() throws JsonProcessingException {
+ @Override
+ protected Class<ClusterConfigurationInfo> getTestResponseClass() {
+ return ClusterConfigurationInfo.class;
+ }
+
+ @Override
+ protected ClusterConfigurationInfo getTestResponseInstance() {
final ClusterConfigurationInfo expected = new ClusterConfigurationInfo(2);
expected.add(new ClusterConfigurationInfoEntry("key1", "value1"));
expected.add(new ClusterConfigurationInfoEntry("key2", "value2"));
- final ObjectMapper objectMapper = RestMapperUtils.getStrictObjectMapper();
-
- JsonNode marshaled = objectMapper.valueToTree(expected);
-
- final ClusterConfigurationInfo unmarshaled = objectMapper.treeToValue(marshaled, ClusterConfigurationInfo.class);
-
- assertEquals(expected, unmarshaled);
+ return expected;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4ba3eecd/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/DashboardConfigurationTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/DashboardConfigurationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/DashboardConfigurationTest.java
index 9a9046b..bb1a6ec 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/DashboardConfigurationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/DashboardConfigurationTest.java
@@ -18,39 +18,23 @@
package org.apache.flink.runtime.rest.handler.legacy.messages;
-import org.apache.flink.runtime.rest.util.RestMapperUtils;
-import org.apache.flink.util.TestLogger;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-
/**
* Tests for the {@link DashboardConfiguration}.
*/
-public class DashboardConfigurationTest extends TestLogger {
+public class DashboardConfigurationTest extends RestResponseMarshallingTestBase<DashboardConfiguration> {
+
+ @Override
+ protected Class<DashboardConfiguration> getTestResponseClass() {
+ return DashboardConfiguration.class;
+ }
- /**
- * Tests that we can marshal and unmarshal {@link DashboardConfiguration} objects.
- */
- @Test
- public void testJsonMarshalling() throws JsonProcessingException {
- final DashboardConfiguration expected = new DashboardConfiguration(
+ @Override
+ protected DashboardConfiguration getTestResponseInstance() {
+ return new DashboardConfiguration(
1L,
"foobar",
42,
"version",
"revision");
-
- final ObjectMapper objectMapper = RestMapperUtils.getStrictObjectMapper();
-
- JsonNode marshaled = objectMapper.valueToTree(expected);
-
- final DashboardConfiguration unmarshaled = objectMapper.treeToValue(marshaled, DashboardConfiguration.class);
-
- assertEquals(expected, unmarshaled);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4ba3eecd/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/RestResponseMarshallingTestBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/RestResponseMarshallingTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/RestResponseMarshallingTestBase.java
new file mode 100644
index 0000000..fce60ac
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/RestResponseMarshallingTestBase.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.legacy.messages;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.util.RestMapperUtils;
+import org.apache.flink.util.TestLogger;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Test base for verifying that marshalling / unmarshalling REST {@link ResponseBody}s work properly.
+ */
+public abstract class RestResponseMarshallingTestBase<R extends ResponseBody> extends TestLogger {
+
+ /**
+ * Returns the class of the test response.
+ *
+ * @return class of the test response type
+ */
+ protected abstract Class<R> getTestResponseClass();
+
+ /**
+ * Returns an instance of a response to be tested.
+ *
+ * @return instance of the expected test response
+ */
+ protected abstract R getTestResponseInstance();
+
+ /**
+ * Tests that we can marshal and unmarshal the response.
+ */
+ @Test
+ public void testJsonMarshalling() throws JsonProcessingException {
+ final R expected = getTestResponseInstance();
+
+ ObjectMapper objectMapper = RestMapperUtils.getStrictObjectMapper();
+ JsonNode json = objectMapper.valueToTree(expected);
+
+ final R unmarshalled = objectMapper.treeToValue(json, getTestResponseClass());
+ Assert.assertEquals(expected, unmarshalled);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/4ba3eecd/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/StatusOverviewWithVersionTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/StatusOverviewWithVersionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/StatusOverviewWithVersionTest.java
index a1bbc9a..6b01dbe 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/StatusOverviewWithVersionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/messages/StatusOverviewWithVersionTest.java
@@ -18,27 +18,19 @@
package org.apache.flink.runtime.rest.handler.legacy.messages;
-import org.apache.flink.runtime.rest.util.RestMapperUtils;
-import org.apache.flink.util.TestLogger;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-
/**
* Tests for the {@link StatusOverviewWithVersion}.
*/
-public class StatusOverviewWithVersionTest extends TestLogger {
+public class StatusOverviewWithVersionTest extends RestResponseMarshallingTestBase<StatusOverviewWithVersion> {
+
+ @Override
+ protected Class<StatusOverviewWithVersion> getTestResponseClass() {
+ return StatusOverviewWithVersion.class;
+ }
- /**
- * Tests that we can marshal and unmarshal StatusOverviewWithVersion.
- */
- @Test
- public void testJsonMarshalling() throws JsonProcessingException {
- final StatusOverviewWithVersion expected = new StatusOverviewWithVersion(
+ @Override
+ protected StatusOverviewWithVersion getTestResponseInstance() {
+ return new StatusOverviewWithVersion(
1,
3,
3,
@@ -48,13 +40,5 @@ public class StatusOverviewWithVersionTest extends TestLogger {
0,
"version",
"commit");
-
- ObjectMapper objectMapper = RestMapperUtils.getStrictObjectMapper();
-
- JsonNode json = objectMapper.valueToTree(expected);
-
- final StatusOverviewWithVersion unmarshalled = objectMapper.treeToValue(json, StatusOverviewWithVersion.class);
-
- assertEquals(expected, unmarshalled);
}
}
[8/9] flink git commit: hotfix] [docs] Fix incorrect doc for s3a
configuration for AWS deployment
Posted by tz...@apache.org.
hotfix] [docs] Fix incorrect doc for s3a configuration for AWS deployment
This closes #4679.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b0548418
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b0548418
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b0548418
Branch: refs/heads/master
Commit: b05484188f2a9ef99b3a80e5d7686eea2df0218d
Parents: bd0841e
Author: Tony Wei <to...@gmail.com>
Authored: Wed Aug 30 11:44:27 2017 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Tue Sep 26 18:04:07 2017 +0200
----------------------------------------------------------------------
docs/ops/deployment/aws.md | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b0548418/docs/ops/deployment/aws.md
----------------------------------------------------------------------
diff --git a/docs/ops/deployment/aws.md b/docs/ops/deployment/aws.md
index 57c7e58..9c6e302 100644
--- a/docs/ops/deployment/aws.md
+++ b/docs/ops/deployment/aws.md
@@ -117,14 +117,14 @@ You need to point Flink to a valid Hadoop configuration, which contains the foll
<!-- Comma separated list of local directories used to buffer
large results prior to transmitting them to S3. -->
<property>
- <name>fs.s3.buffer.dir</name>
+ <name>fs.s3a.buffer.dir</name>
<value>/tmp</value>
</property>
</configuration>
```
-This registers `S3AFileSystem` as the default FileSystem for URIs with the `s3://` scheme.
+This registers `S3AFileSystem` as the default FileSystem for URIs with the `s3a://` scheme.
#### `NativeS3FileSystem`
[4/9] flink git commit: [hotfix] [docs] Method name for
PatternFlatSelectFunction should be flatSelect
Posted by tz...@apache.org.
[hotfix] [docs] Method name for PatternFlatSelectFunction should be flatSelect
This closes #4687.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/415e7d03
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/415e7d03
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/415e7d03
Branch: refs/heads/master
Commit: 415e7d03e655b93c9c0e398f7ea00bc3c7dc440d
Parents: 2b82578
Author: David Anderson <da...@alpinegizmo.com>
Authored: Wed Sep 20 15:57:40 2017 +0200
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Tue Sep 26 18:04:07 2017 +0200
----------------------------------------------------------------------
docs/dev/libs/cep.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/415e7d03/docs/dev/libs/cep.md
----------------------------------------------------------------------
diff --git a/docs/dev/libs/cep.md b/docs/dev/libs/cep.md
index 492f95f..8b16ac4 100644
--- a/docs/dev/libs/cep.md
+++ b/docs/dev/libs/cep.md
@@ -1416,7 +1416,7 @@ used to forward your output elements downstream.
{% highlight java %}
class MyPatternFlatSelectFunction<IN, OUT> implements PatternFlatSelectFunction<IN, OUT> {
@Override
- public void select(Map<String, List<IN>> pattern, Collector<OUT> collector) {
+ public void flatSelect(Map<String, List<IN>> pattern, Collector<OUT> collector) {
IN startEvent = pattern.get("start").get(0);
IN endEvent = pattern.get("end").get(0);