http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/ShardConsumer.java ---------------------------------------------------------------------- diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/ShardConsumer.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/ShardConsumer.java new file mode 100644 index 0000000..612a4a7 --- /dev/null +++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/ShardConsumer.java @@ -0,0 +1,287 @@ +/* + * 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.kinesis.internals; + +import com.amazonaws.services.kinesis.clientlibrary.types.UserRecord; +import com.amazonaws.services.kinesis.model.ExpiredIteratorException; +import com.amazonaws.services.kinesis.model.GetRecordsResult; +import com.amazonaws.services.kinesis.model.Record; +import com.amazonaws.services.kinesis.model.ShardIteratorType; +import org.apache.flink.streaming.api.TimeCharacteristic; +import org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants; +import org.apache.flink.streaming.connectors.kinesis.model.KinesisStreamShard; +import org.apache.flink.streaming.connectors.kinesis.model.SentinelSequenceNumber; +import org.apache.flink.streaming.connectors.kinesis.model.SequenceNumber; +import org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxyInterface; +import org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxy; +import org.apache.flink.streaming.connectors.kinesis.serialization.KinesisDeserializationSchema; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.math.BigInteger; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Properties; + +import static org.apache.flink.util.Preconditions.checkArgument; +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * Thread that does the actual data pulling from AWS Kinesis shards. Each thread is in charge of one Kinesis shard only. + */ +public class ShardConsumer<T> implements Runnable { + + private static final Logger LOG = LoggerFactory.getLogger(ShardConsumer.class); + + private final KinesisDeserializationSchema<T> deserializer; + + private final KinesisProxyInterface kinesis; + + private final int subscribedShardStateIndex; + + private final KinesisDataFetcher<T> fetcherRef; + + private final KinesisStreamShard subscribedShard; + + private final int maxNumberOfRecordsPerFetch; + private final long fetchIntervalMillis; + + private SequenceNumber lastSequenceNum; + + /** + * Creates a shard consumer. + * + * @param fetcherRef reference to the owning fetcher + * @param subscribedShardStateIndex the state index of the shard this consumer is subscribed to + * @param subscribedShard the shard this consumer is subscribed to + * @param lastSequenceNum the sequence number in the shard to start consuming + */ + public ShardConsumer(KinesisDataFetcher<T> fetcherRef, + Integer subscribedShardStateIndex, + KinesisStreamShard subscribedShard, + SequenceNumber lastSequenceNum) { + this(fetcherRef, + subscribedShardStateIndex, + subscribedShard, + lastSequenceNum, + KinesisProxy.create(fetcherRef.getConsumerConfiguration())); + } + + /** This constructor is exposed for testing purposes */ + protected ShardConsumer(KinesisDataFetcher<T> fetcherRef, + Integer subscribedShardStateIndex, + KinesisStreamShard subscribedShard, + SequenceNumber lastSequenceNum, + KinesisProxyInterface kinesis) { + this.fetcherRef = checkNotNull(fetcherRef); + this.subscribedShardStateIndex = checkNotNull(subscribedShardStateIndex); + this.subscribedShard = checkNotNull(subscribedShard); + this.lastSequenceNum = checkNotNull(lastSequenceNum); + checkArgument( + !lastSequenceNum.equals(SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get()), + "Should not start a ShardConsumer if the shard has already been completely read."); + + this.deserializer = fetcherRef.getClonedDeserializationSchema(); + + Properties consumerConfig = fetcherRef.getConsumerConfiguration(); + this.kinesis = kinesis; + this.maxNumberOfRecordsPerFetch = Integer.valueOf(consumerConfig.getProperty( + ConsumerConfigConstants.SHARD_GETRECORDS_MAX, + Integer.toString(ConsumerConfigConstants.DEFAULT_SHARD_GETRECORDS_MAX))); + this.fetchIntervalMillis = Long.valueOf(consumerConfig.getProperty( + ConsumerConfigConstants.SHARD_GETRECORDS_INTERVAL_MILLIS, + Long.toString(ConsumerConfigConstants.DEFAULT_SHARD_GETRECORDS_INTERVAL_MILLIS))); + } + + @SuppressWarnings("unchecked") + @Override + public void run() { + String nextShardItr; + + try { + // before infinitely looping, we set the initial nextShardItr appropriately + + if (lastSequenceNum.equals(SentinelSequenceNumber.SENTINEL_LATEST_SEQUENCE_NUM.get())) { + // if the shard is already closed, there will be no latest next record to get for this shard + if (subscribedShard.isClosed()) { + nextShardItr = null; + } else { + nextShardItr = kinesis.getShardIterator(subscribedShard, ShardIteratorType.LATEST.toString(), null); + } + } else if (lastSequenceNum.equals(SentinelSequenceNumber.SENTINEL_EARLIEST_SEQUENCE_NUM.get())) { + nextShardItr = kinesis.getShardIterator(subscribedShard, ShardIteratorType.TRIM_HORIZON.toString(), null); + } else if (lastSequenceNum.equals(SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get())) { + nextShardItr = null; + } else { + // we will be starting from an actual sequence number (due to restore from failure). + // if the last sequence number refers to an aggregated record, we need to clean up any dangling sub-records + // from the last aggregated record; otherwise, we can simply start iterating from the record right after. + + if (lastSequenceNum.isAggregated()) { + String itrForLastAggregatedRecord = + kinesis.getShardIterator(subscribedShard, ShardIteratorType.AT_SEQUENCE_NUMBER.toString(), lastSequenceNum.getSequenceNumber()); + + // get only the last aggregated record + GetRecordsResult getRecordsResult = getRecords(itrForLastAggregatedRecord, 1); + + List<UserRecord> fetchedRecords = deaggregateRecords( + getRecordsResult.getRecords(), + subscribedShard.getShard().getHashKeyRange().getStartingHashKey(), + subscribedShard.getShard().getHashKeyRange().getEndingHashKey()); + + long lastSubSequenceNum = lastSequenceNum.getSubSequenceNumber(); + for (UserRecord record : fetchedRecords) { + // we have found a dangling sub-record if it has a larger subsequence number + // than our last sequence number; if so, collect the record and update state + if (record.getSubSequenceNumber() > lastSubSequenceNum) { + deserializeRecordForCollectionAndUpdateState(record); + } + } + + // set the nextShardItr so we can continue iterating in the next while loop + nextShardItr = getRecordsResult.getNextShardIterator(); + } else { + // the last record was non-aggregated, so we can simply start from the next record + nextShardItr = kinesis.getShardIterator(subscribedShard, ShardIteratorType.AFTER_SEQUENCE_NUMBER.toString(), lastSequenceNum.getSequenceNumber()); + } + } + + while(isRunning()) { + if (nextShardItr == null) { + fetcherRef.updateState(subscribedShardStateIndex, SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get()); + + // we can close this consumer thread once we've reached the end of the subscribed shard + break; + } else { + if (fetchIntervalMillis != 0) { + Thread.sleep(fetchIntervalMillis); + } + + GetRecordsResult getRecordsResult = getRecords(nextShardItr, maxNumberOfRecordsPerFetch); + + // each of the Kinesis records may be aggregated, so we must deaggregate them before proceeding + List<UserRecord> fetchedRecords = deaggregateRecords( + getRecordsResult.getRecords(), + subscribedShard.getShard().getHashKeyRange().getStartingHashKey(), + subscribedShard.getShard().getHashKeyRange().getEndingHashKey()); + + for (UserRecord record : fetchedRecords) { + deserializeRecordForCollectionAndUpdateState(record); + } + + nextShardItr = getRecordsResult.getNextShardIterator(); + } + } + } catch (Throwable t) { + fetcherRef.stopWithError(t); + } + } + + /** + * The loop in run() checks this before fetching next batch of records. Since this runnable will be executed + * by the ExecutorService {@link KinesisDataFetcher#shardConsumersExecutor}, the only way to close down this thread + * would be by calling shutdownNow() on {@link KinesisDataFetcher#shardConsumersExecutor} and let the executor service + * interrupt all currently running {@link ShardConsumer}s. + */ + private boolean isRunning() { + return !Thread.interrupted(); + } + + /** + * Deserializes a record for collection, and accordingly updates the shard state in the fetcher. The last + * successfully collected sequence number in this shard consumer is also updated so that + * {@link ShardConsumer#getRecords(String, int)} may be able to use the correct sequence number to refresh shard + * iterators if necessary. + * + * Note that the server-side Kinesis timestamp is attached to the record when collected. When the + * user programs uses {@link TimeCharacteristic#EventTime}, this timestamp will be used by default. + * + * @param record record to deserialize and collect + * @throws IOException + */ + private void deserializeRecordForCollectionAndUpdateState(UserRecord record) + throws IOException { + ByteBuffer recordData = record.getData(); + + byte[] dataBytes = new byte[recordData.remaining()]; + recordData.get(dataBytes); + + final long approxArrivalTimestamp = record.getApproximateArrivalTimestamp().getTime(); + + final T value = deserializer.deserialize( + dataBytes, + record.getPartitionKey(), + record.getSequenceNumber(), + approxArrivalTimestamp, + subscribedShard.getStreamName(), + subscribedShard.getShard().getShardId()); + + SequenceNumber collectedSequenceNumber = (record.isAggregated()) + ? new SequenceNumber(record.getSequenceNumber(), record.getSubSequenceNumber()) + : new SequenceNumber(record.getSequenceNumber()); + + fetcherRef.emitRecordAndUpdateState( + value, + approxArrivalTimestamp, + subscribedShardStateIndex, + collectedSequenceNumber); + + lastSequenceNum = collectedSequenceNumber; + } + + /** + * Calls {@link KinesisProxyInterface#getRecords(String, int)}, while also handling unexpected + * AWS {@link ExpiredIteratorException}s to assure that we get results and don't just fail on + * such occasions. The returned shard iterator within the successful {@link GetRecordsResult} should + * be used for the next call to this method. + * + * Note: it is important that this method is not called again before all the records from the last result have been + * fully collected with {@link ShardConsumer#deserializeRecordForCollectionAndUpdateState(UserRecord)}, otherwise + * {@link ShardConsumer#lastSequenceNum} may refer to a sub-record in the middle of an aggregated record, leading to + * incorrect shard iteration if the iterator had to be refreshed. + * + * @param shardItr shard iterator to use + * @param maxNumberOfRecords the maximum number of records to fetch for this getRecords attempt + * @return get records result + * @throws InterruptedException + */ + private GetRecordsResult getRecords(String shardItr, int maxNumberOfRecords) throws InterruptedException { + GetRecordsResult getRecordsResult = null; + while (getRecordsResult == null) { + try { + getRecordsResult = kinesis.getRecords(shardItr, maxNumberOfRecords); + } catch (ExpiredIteratorException eiEx) { + LOG.warn("Encountered an unexpected expired iterator {} for shard {};" + + " refreshing the iterator ...", shardItr, subscribedShard); + shardItr = kinesis.getShardIterator(subscribedShard, ShardIteratorType.AFTER_SEQUENCE_NUMBER.toString(), lastSequenceNum.getSequenceNumber()); + + // sleep for the fetch interval before the next getRecords attempt with the refreshed iterator + if (fetchIntervalMillis != 0) { + Thread.sleep(fetchIntervalMillis); + } + } + } + return getRecordsResult; + } + + @SuppressWarnings("unchecked") + protected static List<UserRecord> deaggregateRecords(List<Record> records, String startingHashKey, String endingHashKey) { + return UserRecord.deaggregate(records, new BigInteger(startingHashKey), new BigInteger(endingHashKey)); + } +}
http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/KinesisStreamShard.java ---------------------------------------------------------------------- diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/KinesisStreamShard.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/KinesisStreamShard.java new file mode 100644 index 0000000..53ed11b --- /dev/null +++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/KinesisStreamShard.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.kinesis.model; + +import com.amazonaws.services.kinesis.model.Shard; + +import java.io.Serializable; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * A serializable representation of a AWS Kinesis Stream shard. It is basically a wrapper class around the information + * provided along with {@link com.amazonaws.services.kinesis.model.Shard}, with some extra utility methods to + * determine whether or not a shard is closed and whether or not the shard is a result of parent shard splits or merges. + */ +public class KinesisStreamShard implements Serializable { + + private static final long serialVersionUID = -6004217801761077536L; + + private final String streamName; + private final Shard shard; + + private final int cachedHash; + + /** + * Create a new KinesisStreamShard + * + * @param streamName + * the name of the Kinesis stream that this shard belongs to + * @param shard + * the actual AWS Shard instance that will be wrapped within this KinesisStreamShard + */ + public KinesisStreamShard(String streamName, Shard shard) { + this.streamName = checkNotNull(streamName); + this.shard = checkNotNull(shard); + + // since our description of Kinesis Streams shards can be fully defined with the stream name and shard id, + // our hash doesn't need to use hash code of Amazon's description of Shards, which uses other info for calculation + int hash = 17; + hash = 37 * hash + streamName.hashCode(); + hash = 37 * hash + shard.getShardId().hashCode(); + this.cachedHash = hash; + } + + public String getStreamName() { + return streamName; + } + + public boolean isClosed() { + return (shard.getSequenceNumberRange().getEndingSequenceNumber() != null); + } + + public Shard getShard() { + return shard; + } + + @Override + public String toString() { + return "KinesisStreamShard{" + + "streamName='" + streamName + "'" + + ", shard='" + shard.toString() + "'}"; + } + + @Override + public boolean equals(Object obj) { + if (!(obj instanceof KinesisStreamShard)) { + return false; + } + + if (obj == this) { + return true; + } + + KinesisStreamShard other = (KinesisStreamShard) obj; + + return streamName.equals(other.getStreamName()) && shard.equals(other.getShard()); + } + + @Override + public int hashCode() { + return cachedHash; + } + + /** + * Utility function to compare two shard ids + * + * @param firstShardId first shard id to compare + * @param secondShardId second shard id to compare + * @return a value less than 0 if the first shard id is smaller than the second shard id, + * or a value larger than 0 the first shard is larger then the second shard id, + * or 0 if they are equal + */ + public static int compareShardIds(String firstShardId, String secondShardId) { + if (!isValidShardId(firstShardId)) { + throw new IllegalArgumentException("The first shard id has invalid format."); + } + + if (!isValidShardId(secondShardId)) { + throw new IllegalArgumentException("The second shard id has invalid format."); + } + + // digit segment of the shard id starts at index 8 + return Long.compare(Long.parseLong(firstShardId.substring(8)), Long.parseLong(secondShardId.substring(8))); + } + + /** + * Checks if a shard id has valid format. + * Kinesis stream shard ids have 12-digit numbers left-padded with 0's, + * prefixed with "shardId-", ex. "shardId-000000000015". + * + * @param shardId the shard id to check + * @return whether the shard id is valid + */ + public static boolean isValidShardId(String shardId) { + if (shardId == null) { return false; } + return shardId.matches("^shardId-\\d{12}"); + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/KinesisStreamShardState.java ---------------------------------------------------------------------- diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/KinesisStreamShardState.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/KinesisStreamShardState.java new file mode 100644 index 0000000..00181da --- /dev/null +++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/KinesisStreamShardState.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.streaming.connectors.kinesis.model; + +/** + * A wrapper class that bundles a {@link KinesisStreamShard} with its last processed sequence number. + */ +public class KinesisStreamShardState { + + private KinesisStreamShard kinesisStreamShard; + private SequenceNumber lastProcessedSequenceNum; + + public KinesisStreamShardState(KinesisStreamShard kinesisStreamShard, SequenceNumber lastProcessedSequenceNum) { + this.kinesisStreamShard = kinesisStreamShard; + this.lastProcessedSequenceNum = lastProcessedSequenceNum; + } + + public KinesisStreamShard getKinesisStreamShard() { + return this.kinesisStreamShard; + } + + public SequenceNumber getLastProcessedSequenceNum() { + return this.lastProcessedSequenceNum; + } + + public void setLastProcessedSequenceNum(SequenceNumber update) { + this.lastProcessedSequenceNum = update; + } + + @Override + public String toString() { + return "KinesisStreamShardState{" + + "kinesisStreamShard='" + kinesisStreamShard.toString() + "'" + + ", lastProcessedSequenceNumber='" + lastProcessedSequenceNum.toString() + "'}"; + } + + @Override + public boolean equals(Object obj) { + if (!(obj instanceof KinesisStreamShardState)) { + return false; + } + + if (obj == this) { + return true; + } + + KinesisStreamShardState other = (KinesisStreamShardState) obj; + + return kinesisStreamShard.equals(other.getKinesisStreamShard()) && lastProcessedSequenceNum.equals(other.getLastProcessedSequenceNum()); + } + + @Override + public int hashCode() { + return 37 * (kinesisStreamShard.hashCode() + lastProcessedSequenceNum.hashCode()); + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/SentinelSequenceNumber.java ---------------------------------------------------------------------- diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/SentinelSequenceNumber.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/SentinelSequenceNumber.java new file mode 100644 index 0000000..8182201 --- /dev/null +++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/SentinelSequenceNumber.java @@ -0,0 +1,51 @@ +/* + * 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.kinesis.model; + +import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer; +import org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher; + +/** + * Special flag values for sequence numbers in shards to indicate special positions. + * The value is initially set by {@link FlinkKinesisConsumer} when {@link KinesisDataFetcher}s are created. + * The KinesisDataFetchers will use this value to determine how to retrieve the starting shard iterator from AWS Kinesis. + */ +public enum SentinelSequenceNumber { + + /** Flag value for shard's sequence numbers to indicate that the + * shard should start to be read from the latest incoming records */ + SENTINEL_LATEST_SEQUENCE_NUM( new SequenceNumber("LATEST_SEQUENCE_NUM") ), + + /** Flag value for shard's sequence numbers to indicate that the shard should + * start to be read from the earliest records that haven't expired yet */ + SENTINEL_EARLIEST_SEQUENCE_NUM( new SequenceNumber("EARLIEST_SEQUENCE_NUM") ), + + /** Flag value to indicate that we have already read the last record of this shard + * (Note: Kinesis shards that have been closed due to a split or merge will have an ending data record) */ + SENTINEL_SHARD_ENDING_SEQUENCE_NUM( new SequenceNumber("SHARD_ENDING_SEQUENCE_NUM") ); + + private SequenceNumber sentinel; + + SentinelSequenceNumber(SequenceNumber sentinel) { + this.sentinel = sentinel; + } + + public SequenceNumber get() { + return sentinel; + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/SequenceNumber.java ---------------------------------------------------------------------- diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/SequenceNumber.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/SequenceNumber.java new file mode 100644 index 0000000..021f53f --- /dev/null +++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/SequenceNumber.java @@ -0,0 +1,104 @@ +/* + * 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.kinesis.model; + +import java.io.Serializable; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * A serializable representation of a Kinesis record's sequence number. It has two fields: the main sequence number, + * and also a subsequence number. If this {@link SequenceNumber} is referring to an aggregated Kinesis record, the + * subsequence number will be a non-negative value representing the order of the sub-record within the aggregation. + */ +public class SequenceNumber implements Serializable { + + private static final long serialVersionUID = 876972197938972667L; + + private static final String DELIMITER = "-"; + + private final String sequenceNumber; + private final long subSequenceNumber; + + private final int cachedHash; + + /** + * Create a new instance for a non-aggregated Kinesis record without a subsequence number. + * @param sequenceNumber the sequence number + */ + public SequenceNumber(String sequenceNumber) { + this(sequenceNumber, -1); + } + + /** + * Create a new instance, with the specified sequence number and subsequence number. + * To represent the sequence number for a non-aggregated Kinesis record, the subsequence number should be -1. + * Otherwise, give a non-negative sequence number to represent an aggregated Kinesis record. + * + * @param sequenceNumber the sequence number + * @param subSequenceNumber the subsequence number (-1 to represent non-aggregated Kinesis records) + */ + public SequenceNumber(String sequenceNumber, long subSequenceNumber) { + this.sequenceNumber = checkNotNull(sequenceNumber); + this.subSequenceNumber = subSequenceNumber; + + this.cachedHash = 37 * (sequenceNumber.hashCode() + Long.valueOf(subSequenceNumber).hashCode()); + } + + public boolean isAggregated() { + return subSequenceNumber >= 0; + } + + public String getSequenceNumber() { + return sequenceNumber; + } + + public long getSubSequenceNumber() { + return subSequenceNumber; + } + + @Override + public String toString() { + if (isAggregated()) { + return sequenceNumber + DELIMITER + subSequenceNumber; + } else { + return sequenceNumber; + } + } + + @Override + public boolean equals(Object obj) { + if (!(obj instanceof SequenceNumber)) { + return false; + } + + if (obj == this) { + return true; + } + + SequenceNumber other = (SequenceNumber) obj; + + return sequenceNumber.equals(other.getSequenceNumber()) + && (subSequenceNumber == other.getSubSequenceNumber()); + } + + @Override + public int hashCode() { + return cachedHash; + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/GetShardListResult.java ---------------------------------------------------------------------- diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/GetShardListResult.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/GetShardListResult.java new file mode 100644 index 0000000..04b1654 --- /dev/null +++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/GetShardListResult.java @@ -0,0 +1,75 @@ +/* + * 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.kinesis.proxy; + +import org.apache.flink.streaming.connectors.kinesis.model.KinesisStreamShard; + +import java.util.LinkedList; +import java.util.List; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; + +/** + * Basic model class to bundle the shards retrieved from Kinesis on a {@link KinesisProxyInterface#getShardList(Map)} call. + */ +public class GetShardListResult { + + private final Map<String, LinkedList<KinesisStreamShard>> streamsToRetrievedShardList = new HashMap<>(); + + public void addRetrievedShardToStream(String stream, KinesisStreamShard retrievedShard) { + if (!streamsToRetrievedShardList.containsKey(stream)) { + streamsToRetrievedShardList.put(stream, new LinkedList<KinesisStreamShard>()); + } + streamsToRetrievedShardList.get(stream).add(retrievedShard); + } + + public void addRetrievedShardsToStream(String stream, List<KinesisStreamShard> retrievedShards) { + if (retrievedShards.size() != 0) { + if (!streamsToRetrievedShardList.containsKey(stream)) { + streamsToRetrievedShardList.put(stream, new LinkedList<KinesisStreamShard>()); + } + streamsToRetrievedShardList.get(stream).addAll(retrievedShards); + } + } + + public List<KinesisStreamShard> getRetrievedShardListOfStream(String stream) { + if (!streamsToRetrievedShardList.containsKey(stream)) { + return null; + } else { + return streamsToRetrievedShardList.get(stream); + } + } + + public KinesisStreamShard getLastSeenShardOfStream(String stream) { + if (!streamsToRetrievedShardList.containsKey(stream)) { + return null; + } else { + return streamsToRetrievedShardList.get(stream).getLast(); + } + } + + public boolean hasRetrievedShards() { + return !streamsToRetrievedShardList.isEmpty(); + } + + public Set<String> getStreamsWithRetrievedShards() { + return streamsToRetrievedShardList.keySet(); + } + +} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java ---------------------------------------------------------------------- diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java new file mode 100644 index 0000000..9ffc8e6 --- /dev/null +++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java @@ -0,0 +1,338 @@ +/* + * 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.kinesis.proxy; + +import com.amazonaws.services.kinesis.AmazonKinesisClient; +import com.amazonaws.services.kinesis.model.DescribeStreamRequest; +import com.amazonaws.services.kinesis.model.DescribeStreamResult; +import com.amazonaws.services.kinesis.model.GetRecordsRequest; +import com.amazonaws.services.kinesis.model.GetRecordsResult; +import com.amazonaws.services.kinesis.model.GetShardIteratorResult; +import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException; +import com.amazonaws.services.kinesis.model.LimitExceededException; +import com.amazonaws.services.kinesis.model.ResourceNotFoundException; +import com.amazonaws.services.kinesis.model.StreamStatus; +import com.amazonaws.services.kinesis.model.Shard; +import org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants; +import org.apache.flink.streaming.connectors.kinesis.model.KinesisStreamShard; +import org.apache.flink.streaming.connectors.kinesis.util.AWSUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Properties; +import java.util.Map; +import java.util.Random; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * Kinesis proxy implementation - a utility class that is used as a proxy to make + * calls to AWS Kinesis for several functions, such as getting a list of shards and + * fetching a batch of data records starting from a specified record sequence number. + * + * NOTE: + * In the AWS KCL library, there is a similar implementation - {@link com.amazonaws.services.kinesis.clientlibrary.proxies.KinesisProxy}. + * This implementation differs mainly in that we can make operations to arbitrary Kinesis streams, which is a needed + * functionality for the Flink Kinesis Connecter since the consumer may simultaneously read from multiple Kinesis streams. + */ +public class KinesisProxy implements KinesisProxyInterface { + + private static final Logger LOG = LoggerFactory.getLogger(KinesisProxy.class); + + /** The actual Kinesis client from the AWS SDK that we will be using to make calls */ + private final AmazonKinesisClient kinesisClient; + + /** Random seed used to calculate backoff jitter for Kinesis operations */ + private final static Random seed = new Random(); + + // ------------------------------------------------------------------------ + // describeStream() related performance settings + // ------------------------------------------------------------------------ + + /** Base backoff millis for the describe stream operation */ + private final long describeStreamBaseBackoffMillis; + + /** Maximum backoff millis for the describe stream operation */ + private final long describeStreamMaxBackoffMillis; + + /** Exponential backoff power constant for the describe stream operation */ + private final double describeStreamExpConstant; + + // ------------------------------------------------------------------------ + // getRecords() related performance settings + // ------------------------------------------------------------------------ + + /** Base backoff millis for the get records operation */ + private final long getRecordsBaseBackoffMillis; + + /** Maximum backoff millis for the get records operation */ + private final long getRecordsMaxBackoffMillis; + + /** Exponential backoff power constant for the get records operation */ + private final double getRecordsExpConstant; + + /** Maximum attempts for the get records operation */ + private final int getRecordsMaxAttempts; + + // ------------------------------------------------------------------------ + // getShardIterator() related performance settings + // ------------------------------------------------------------------------ + + /** Base backoff millis for the get shard iterator operation */ + private final long getShardIteratorBaseBackoffMillis; + + /** Maximum backoff millis for the get shard iterator operation */ + private final long getShardIteratorMaxBackoffMillis; + + /** Exponential backoff power constant for the get shard iterator operation */ + private final double getShardIteratorExpConstant; + + /** Maximum attempts for the get shard iterator operation */ + private final int getShardIteratorMaxAttempts; + + /** + * Create a new KinesisProxy based on the supplied configuration properties + * + * @param configProps configuration properties containing AWS credential and AWS region info + */ + private KinesisProxy(Properties configProps) { + checkNotNull(configProps); + + this.kinesisClient = AWSUtil.createKinesisClient(configProps); + + this.describeStreamBaseBackoffMillis = Long.valueOf( + configProps.getProperty( + ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_BASE, + Long.toString(ConsumerConfigConstants.DEFAULT_STREAM_DESCRIBE_BACKOFF_BASE))); + this.describeStreamMaxBackoffMillis = Long.valueOf( + configProps.getProperty( + ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_MAX, + Long.toString(ConsumerConfigConstants.DEFAULT_STREAM_DESCRIBE_BACKOFF_MAX))); + this.describeStreamExpConstant = Double.valueOf( + configProps.getProperty( + ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_EXPONENTIAL_CONSTANT, + Double.toString(ConsumerConfigConstants.DEFAULT_STREAM_DESCRIBE_BACKOFF_EXPONENTIAL_CONSTANT))); + + this.getRecordsBaseBackoffMillis = Long.valueOf( + configProps.getProperty( + ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_BASE, + Long.toString(ConsumerConfigConstants.DEFAULT_SHARD_GETRECORDS_BACKOFF_BASE))); + this.getRecordsMaxBackoffMillis = Long.valueOf( + configProps.getProperty( + ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_MAX, + Long.toString(ConsumerConfigConstants.DEFAULT_SHARD_GETRECORDS_BACKOFF_MAX))); + this.getRecordsExpConstant = Double.valueOf( + configProps.getProperty( + ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_EXPONENTIAL_CONSTANT, + Double.toString(ConsumerConfigConstants.DEFAULT_SHARD_GETRECORDS_BACKOFF_EXPONENTIAL_CONSTANT))); + this.getRecordsMaxAttempts = Integer.valueOf( + configProps.getProperty( + ConsumerConfigConstants.SHARD_GETRECORDS_RETRIES, + Long.toString(ConsumerConfigConstants.DEFAULT_SHARD_GETRECORDS_RETRIES))); + + this.getShardIteratorBaseBackoffMillis = Long.valueOf( + configProps.getProperty( + ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_BASE, + Long.toString(ConsumerConfigConstants.DEFAULT_SHARD_GETITERATOR_BACKOFF_BASE))); + this.getShardIteratorMaxBackoffMillis = Long.valueOf( + configProps.getProperty( + ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_MAX, + Long.toString(ConsumerConfigConstants.DEFAULT_SHARD_GETITERATOR_BACKOFF_MAX))); + this.getShardIteratorExpConstant = Double.valueOf( + configProps.getProperty( + ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_EXPONENTIAL_CONSTANT, + Double.toString(ConsumerConfigConstants.DEFAULT_SHARD_GETITERATOR_BACKOFF_EXPONENTIAL_CONSTANT))); + this.getShardIteratorMaxAttempts = Integer.valueOf( + configProps.getProperty( + ConsumerConfigConstants.SHARD_GETITERATOR_RETRIES, + Long.toString(ConsumerConfigConstants.DEFAULT_SHARD_GETITERATOR_RETRIES))); + + } + + /** + * Creates a Kinesis proxy. + * + * @param configProps configuration properties + * @return the created kinesis proxy + */ + public static KinesisProxyInterface create(Properties configProps) { + return new KinesisProxy(configProps); + } + + /** + * {@inheritDoc} + */ + @Override + public GetRecordsResult getRecords(String shardIterator, int maxRecordsToGet) throws InterruptedException { + final GetRecordsRequest getRecordsRequest = new GetRecordsRequest(); + getRecordsRequest.setShardIterator(shardIterator); + getRecordsRequest.setLimit(maxRecordsToGet); + + GetRecordsResult getRecordsResult = null; + + int attempt = 0; + while (attempt <= getRecordsMaxAttempts && getRecordsResult == null) { + try { + getRecordsResult = kinesisClient.getRecords(getRecordsRequest); + } catch (ProvisionedThroughputExceededException ex) { + long backoffMillis = fullJitterBackoff( + getRecordsBaseBackoffMillis, getRecordsMaxBackoffMillis, getRecordsExpConstant, attempt++); + LOG.warn("Got ProvisionedThroughputExceededException. Backing off for " + + backoffMillis + " millis."); + Thread.sleep(backoffMillis); + } + } + + if (getRecordsResult == null) { + throw new RuntimeException("Rate Exceeded for getRecords operation - all " + getRecordsMaxAttempts + + " retry attempts returned ProvisionedThroughputExceededException."); + } + + return getRecordsResult; + } + + /** + * {@inheritDoc} + */ + @Override + public GetShardListResult getShardList(Map<String, String> streamNamesWithLastSeenShardIds) throws InterruptedException { + GetShardListResult result = new GetShardListResult(); + + for (Map.Entry<String,String> streamNameWithLastSeenShardId : streamNamesWithLastSeenShardIds.entrySet()) { + String stream = streamNameWithLastSeenShardId.getKey(); + String lastSeenShardId = streamNameWithLastSeenShardId.getValue(); + result.addRetrievedShardsToStream(stream, getShardsOfStream(stream, lastSeenShardId)); + } + return result; + } + + /** + * {@inheritDoc} + */ + @Override + public String getShardIterator(KinesisStreamShard shard, String shardIteratorType, @Nullable String startingSeqNum) throws InterruptedException { + GetShardIteratorResult getShardIteratorResult = null; + + int attempt = 0; + while (attempt <= getShardIteratorMaxAttempts && getShardIteratorResult == null) { + try { + getShardIteratorResult = + kinesisClient.getShardIterator(shard.getStreamName(), shard.getShard().getShardId(), shardIteratorType, startingSeqNum); + } catch (ProvisionedThroughputExceededException ex) { + long backoffMillis = fullJitterBackoff( + getShardIteratorBaseBackoffMillis, getShardIteratorMaxBackoffMillis, getShardIteratorExpConstant, attempt++); + LOG.warn("Got ProvisionedThroughputExceededException. Backing off for " + + backoffMillis + " millis."); + Thread.sleep(backoffMillis); + } + } + + if (getShardIteratorResult == null) { + throw new RuntimeException("Rate Exceeded for getShardIterator operation - all " + getShardIteratorMaxAttempts + + " retry attempts returned ProvisionedThroughputExceededException."); + } + return getShardIteratorResult.getShardIterator(); + } + + private List<KinesisStreamShard> getShardsOfStream(String streamName, @Nullable String lastSeenShardId) throws InterruptedException { + List<KinesisStreamShard> shardsOfStream = new ArrayList<>(); + + DescribeStreamResult describeStreamResult; + do { + describeStreamResult = describeStream(streamName, lastSeenShardId); + + List<Shard> shards = describeStreamResult.getStreamDescription().getShards(); + for (Shard shard : shards) { + shardsOfStream.add(new KinesisStreamShard(streamName, shard)); + } + + if (shards.size() != 0) { + lastSeenShardId = shards.get(shards.size() - 1).getShardId(); + } + } while (describeStreamResult.getStreamDescription().isHasMoreShards()); + + return shardsOfStream; + } + + /** + * Get metainfo for a Kinesis stream, which contains information about which shards this Kinesis stream possess. + * + * This method is using a "full jitter" approach described in AWS's article, + * <a href="https://www.awsarchitectureblog.com/2015/03/backoff.html";>"Exponential Backoff and Jitter"</a>. + * This is necessary because concurrent calls will be made by all parallel subtask's fetcher. This + * jitter backoff approach will help distribute calls across the fetchers over time. + * + * @param streamName the stream to describe + * @param startShardId which shard to start with for this describe operation (earlier shard's infos will not appear in result) + * @return the result of the describe stream operation + */ + private DescribeStreamResult describeStream(String streamName, @Nullable String startShardId) throws InterruptedException { + final DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest(); + describeStreamRequest.setStreamName(streamName); + describeStreamRequest.setExclusiveStartShardId(startShardId); + + DescribeStreamResult describeStreamResult = null; + + // Call DescribeStream, with full-jitter backoff (if we get LimitExceededException). + int attemptCount = 0; + while (describeStreamResult == null) { // retry until we get a result + try { + describeStreamResult = kinesisClient.describeStream(describeStreamRequest); + } catch (LimitExceededException le) { + long backoffMillis = fullJitterBackoff( + describeStreamBaseBackoffMillis, describeStreamMaxBackoffMillis, describeStreamExpConstant, attemptCount++); + LOG.warn("Got LimitExceededException when describing stream " + streamName + ". Backing off for " + + backoffMillis + " millis."); + Thread.sleep(backoffMillis); + } catch (ResourceNotFoundException re) { + throw new RuntimeException("Error while getting stream details", re); + } + } + + String streamStatus = describeStreamResult.getStreamDescription().getStreamStatus(); + if (!(streamStatus.equals(StreamStatus.ACTIVE.toString()) || streamStatus.equals(StreamStatus.UPDATING.toString()))) { + if (LOG.isWarnEnabled()) { + LOG.warn("The status of stream " + streamName + " is " + streamStatus + "; result of the current " + + "describeStream operation will not contain any shard information."); + } + } + + // Kinesalite (mock implementation of Kinesis) does not correctly exclude shards before the exclusive + // start shard id in the returned shards list; check if we need to remove these erroneously returned shards + if (startShardId != null) { + List<Shard> shards = describeStreamResult.getStreamDescription().getShards(); + Iterator<Shard> shardItr = shards.iterator(); + while (shardItr.hasNext()) { + if (KinesisStreamShard.compareShardIds(shardItr.next().getShardId(), startShardId) <= 0) { + shardItr.remove(); + } + } + } + + return describeStreamResult; + } + + private static long fullJitterBackoff(long base, long max, double power, int attempt) { + long exponentialBackoff = (long) Math.min(max, base * Math.pow(power, attempt)); + return (long)(seed.nextDouble()*exponentialBackoff); // random jitter between 0 and the exponential backoff + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyInterface.java ---------------------------------------------------------------------- diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyInterface.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyInterface.java new file mode 100644 index 0000000..39ddc52 --- /dev/null +++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyInterface.java @@ -0,0 +1,69 @@ +/* + * 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.kinesis.proxy; + +import com.amazonaws.services.kinesis.model.GetRecordsResult; +import org.apache.flink.streaming.connectors.kinesis.model.KinesisStreamShard; + +import java.util.Map; + +/** + * Interface for a Kinesis proxy that operates on multiple Kinesis streams within the same AWS service region. + */ +public interface KinesisProxyInterface { + + /** + * Get a shard iterator from the specified position in a shard. + * The retrieved shard iterator can be used in {@link KinesisProxyInterface#getRecords(String, int)}} + * to read data from the Kinesis shard. + * + * @param shard the shard to get the iterator + * @param shardIteratorType the iterator type, defining how the shard is to be iterated + * (one of: TRIM_HORIZON, LATEST, AT_SEQUENCE_NUMBER, AFTER_SEQUENCE_NUMBER) + * @param startingSeqNum sequence number, must be null if shardIteratorType is TRIM_HORIZON or LATEST + * @return shard iterator which can be used to read data from Kinesis + * @throws InterruptedException this method will retry with backoff if AWS Kinesis complains that the + * operation has exceeded the rate limit; this exception will be thrown + * if the backoff is interrupted. + */ + String getShardIterator(KinesisStreamShard shard, String shardIteratorType, String startingSeqNum) throws InterruptedException; + + /** + * Get the next batch of data records using a specific shard iterator + * + * @param shardIterator a shard iterator that encodes info about which shard to read and where to start reading + * @param maxRecordsToGet the maximum amount of records to retrieve for this batch + * @return the batch of retrieved records, also with a shard iterator that can be used to get the next batch + * @throws InterruptedException this method will retry with backoff if AWS Kinesis complains that the + * operation has exceeded the rate limit; this exception will be thrown + * if the backoff is interrupted. + */ + GetRecordsResult getRecords(String shardIterator, int maxRecordsToGet) throws InterruptedException; + + /** + * Get shard list of multiple Kinesis streams, ignoring the + * shards of each stream before a specified last seen shard id. + * + * @param streamNamesWithLastSeenShardIds a map with stream as key, and last seen shard id as value + * @return result of the shard list query + * @throws InterruptedException this method will retry with backoff if AWS Kinesis complains that the + * operation has exceeded the rate limit; this exception will be thrown + * if the backoff is interrupted. + */ + GetShardListResult getShardList(Map<String,String> streamNamesWithLastSeenShardIds) throws InterruptedException; +} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/serialization/KinesisDeserializationSchema.java ---------------------------------------------------------------------- diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/serialization/KinesisDeserializationSchema.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/serialization/KinesisDeserializationSchema.java new file mode 100644 index 0000000..0effdd8 --- /dev/null +++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/serialization/KinesisDeserializationSchema.java @@ -0,0 +1,57 @@ +/* + * 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.kinesis.serialization; + +import org.apache.flink.api.java.typeutils.ResultTypeQueryable; +import org.apache.flink.streaming.util.serialization.DeserializationSchema; + +import java.io.IOException; +import java.io.Serializable; + +/** + * This is a deserialization schema specific for the Flink Kinesis Consumer. Different from the + * basic {@link DeserializationSchema}, this schema offers additional Kinesis-specific information + * about the record that may be useful to the user application. + * + * @param <T> The type created by the keyed deserialization schema. + */ +public interface KinesisDeserializationSchema<T> extends Serializable, ResultTypeQueryable<T> { + + /** + * Deserializes a Kinesis record's bytes + * + * @param recordValue the record's value as a byte array + * @param partitionKey the record's partition key at the time of writing + * @param seqNum the sequence number of this record in the Kinesis shard + * @param approxArrivalTimestamp the server-side timestamp of when Kinesis received and stored the record + * @param stream the name of the Kinesis stream that this record was sent to + * @param shardId The identifier of the shard the record was sent to + * @return the deserialized message as an Java object + * @throws IOException + */ + T deserialize(byte[] recordValue, String partitionKey, String seqNum, long approxArrivalTimestamp, String stream, String shardId) throws IOException; + + /** + * Method to decide whether the element signals the end of the stream. If + * true is returned the element won't be emitted. + * + * @param nextElement the element to test for the end-of-stream signal + * @return true if the element signals end of stream, false otherwise + */ + // TODO FLINK-4194 ADD SUPPORT FOR boolean isEndOfStream(T nextElement); +} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/serialization/KinesisDeserializationSchemaWrapper.java ---------------------------------------------------------------------- diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/serialization/KinesisDeserializationSchemaWrapper.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/serialization/KinesisDeserializationSchemaWrapper.java new file mode 100644 index 0000000..6e66038 --- /dev/null +++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/serialization/KinesisDeserializationSchemaWrapper.java @@ -0,0 +1,57 @@ +/* + * 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.kinesis.serialization; + +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.streaming.util.serialization.DeserializationSchema; + +import java.io.IOException; + +/** + * A simple wrapper for using the {@link DeserializationSchema} with the {@link KinesisDeserializationSchema} interface + * + * @param <T> The type created by the deserialization schema. + */ +public class KinesisDeserializationSchemaWrapper<T> implements KinesisDeserializationSchema<T> { + private static final long serialVersionUID = 9143148962928375886L; + + private final DeserializationSchema<T> deserializationSchema; + + public KinesisDeserializationSchemaWrapper(DeserializationSchema<T> deserializationSchema) { + this.deserializationSchema = deserializationSchema; + } + + @Override + public T deserialize(byte[] recordValue, String partitionKey, String seqNum, long approxArrivalTimestamp, String stream, String shardId) + throws IOException { + return deserializationSchema.deserialize(recordValue); + } + + /* + FLINK-4194 + + @Override + public boolean isEndOfStream(T nextElement) { + return deserializationSchema.isEndOfStream(nextElement); + } */ + + @Override + public TypeInformation<T> getProducedType() { + return deserializationSchema.getProducedType(); + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/serialization/KinesisSerializationSchema.java ---------------------------------------------------------------------- diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/serialization/KinesisSerializationSchema.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/serialization/KinesisSerializationSchema.java new file mode 100644 index 0000000..03dd72c --- /dev/null +++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/serialization/KinesisSerializationSchema.java @@ -0,0 +1,45 @@ +/* + * 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.kinesis.serialization; + + +import java.io.Serializable; +import java.nio.ByteBuffer; + +/** + * Kinesis-specific serialization schema, allowing users to specify a target stream based + * on a record's contents. + * @param <T> + */ +public interface KinesisSerializationSchema<T> extends Serializable { + /** + * Serialize the given element into a ByteBuffer + * + * @param element The element to serialize + * @return Serialized representation of the element + */ + ByteBuffer serialize(T element); + + /** + * Optional method to determine the target stream based on the element. + * Return <code>null</code> to use the default stream + * + * @param element The element to determine the target stream from + * @return target stream name + */ + String getTargetStream(T element); +} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AWSUtil.java ---------------------------------------------------------------------- diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AWSUtil.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AWSUtil.java new file mode 100644 index 0000000..cff69e5 --- /dev/null +++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AWSUtil.java @@ -0,0 +1,130 @@ +/* + * 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.kinesis.util; + +import com.amazonaws.ClientConfiguration; +import com.amazonaws.ClientConfigurationFactory; +import com.amazonaws.auth.AWSCredentials; +import com.amazonaws.auth.AWSCredentialsProvider; +import com.amazonaws.auth.BasicAWSCredentials; +import com.amazonaws.auth.DefaultAWSCredentialsProviderChain; +import com.amazonaws.auth.EnvironmentVariableCredentialsProvider; +import com.amazonaws.auth.SystemPropertiesCredentialsProvider; +import com.amazonaws.auth.profile.ProfileCredentialsProvider; +import com.amazonaws.regions.Region; +import com.amazonaws.regions.Regions; +import com.amazonaws.services.kinesis.AmazonKinesisClient; +import org.apache.flink.runtime.util.EnvironmentInformation; +import org.apache.flink.streaming.connectors.kinesis.config.AWSConfigConstants; +import org.apache.flink.streaming.connectors.kinesis.config.AWSConfigConstants.CredentialProvider; + +import java.util.Properties; + +/** + * Some utilities specific to Amazon Web Service. + */ +public class AWSUtil { + + /** + * Creates an Amazon Kinesis Client. + * @param configProps configuration properties containing the access key, secret key, and region + * @return a new Amazon Kinesis Client + */ + public static AmazonKinesisClient createKinesisClient(Properties configProps) { + // set a Flink-specific user agent + ClientConfiguration awsClientConfig = new ClientConfigurationFactory().getConfig(); + awsClientConfig.setUserAgent("Apache Flink " + EnvironmentInformation.getVersion() + + " (" + EnvironmentInformation.getRevisionInformation().commitId + ") Kinesis Connector"); + + // utilize automatic refreshment of credentials by directly passing the AWSCredentialsProvider + AmazonKinesisClient client = new AmazonKinesisClient( + AWSUtil.getCredentialsProvider(configProps), awsClientConfig); + + client.setRegion(Region.getRegion(Regions.fromName(configProps.getProperty(AWSConfigConstants.AWS_REGION)))); + if (configProps.containsKey(AWSConfigConstants.AWS_ENDPOINT)) { + client.setEndpoint(configProps.getProperty(AWSConfigConstants.AWS_ENDPOINT)); + } + return client; + } + + /** + * Return a {@link AWSCredentialsProvider} instance corresponding to the configuration properties. + * + * @param configProps the configuration properties + * @return The corresponding AWS Credentials Provider instance + */ + public static AWSCredentialsProvider getCredentialsProvider(final Properties configProps) { + CredentialProvider credentialProviderType = CredentialProvider.valueOf(configProps.getProperty( + AWSConfigConstants.AWS_CREDENTIALS_PROVIDER, CredentialProvider.BASIC.toString())); + + AWSCredentialsProvider credentialsProvider; + + switch (credentialProviderType) { + case ENV_VAR: + credentialsProvider = new EnvironmentVariableCredentialsProvider(); + break; + case SYS_PROP: + credentialsProvider = new SystemPropertiesCredentialsProvider(); + break; + case PROFILE: + String profileName = configProps.getProperty( + AWSConfigConstants.AWS_PROFILE_NAME, null); + String profileConfigPath = configProps.getProperty( + AWSConfigConstants.AWS_PROFILE_PATH, null); + credentialsProvider = (profileConfigPath == null) + ? new ProfileCredentialsProvider(profileName) + : new ProfileCredentialsProvider(profileConfigPath, profileName); + break; + case AUTO: + credentialsProvider = new DefaultAWSCredentialsProviderChain(); + break; + default: + case BASIC: + credentialsProvider = new AWSCredentialsProvider() { + @Override + public AWSCredentials getCredentials() { + return new BasicAWSCredentials( + configProps.getProperty(AWSConfigConstants.AWS_ACCESS_KEY_ID), + configProps.getProperty(AWSConfigConstants.AWS_SECRET_ACCESS_KEY)); + } + + @Override + public void refresh() { + // do nothing + } + }; + } + + return credentialsProvider; + } + + /** + * Checks whether or not a region ID is valid + * + * @param region The AWS region ID to check + * @return true if the supplied region ID is valid, false otherwise + */ + public static boolean isValidRegion(String region) { + try { + Regions.fromName(region.toLowerCase()); + } catch (IllegalArgumentException e) { + return false; + } + return true; + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtil.java ---------------------------------------------------------------------- diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtil.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtil.java new file mode 100644 index 0000000..9aa14ad --- /dev/null +++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtil.java @@ -0,0 +1,218 @@ +/* + * 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.kinesis.util; + +import com.amazonaws.regions.Regions; +import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer; +import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisProducer; +import org.apache.flink.streaming.connectors.kinesis.config.AWSConfigConstants; +import org.apache.flink.streaming.connectors.kinesis.config.AWSConfigConstants.CredentialProvider; +import org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants; +import org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.InitialPosition; +import org.apache.flink.streaming.connectors.kinesis.config.ProducerConfigConstants; + +import java.util.Properties; + +import static org.apache.flink.util.Preconditions.checkArgument; +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * Utilities for Flink Kinesis connector configuration. + */ +public class KinesisConfigUtil { + + /** + * Validate configuration properties for {@link FlinkKinesisConsumer}. + */ + public static void validateConsumerConfiguration(Properties config) { + checkNotNull(config, "config can not be null"); + + validateAwsConfiguration(config); + + if (config.containsKey(ConsumerConfigConstants.STREAM_INITIAL_POSITION)) { + String initPosType = config.getProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION); + + // specified initial position in stream must be either LATEST or TRIM_HORIZON + try { + InitialPosition.valueOf(initPosType); + } catch (IllegalArgumentException e) { + StringBuilder sb = new StringBuilder(); + for (InitialPosition pos : InitialPosition.values()) { + sb.append(pos.toString()).append(", "); + } + throw new IllegalArgumentException("Invalid initial position in stream set in config. Valid values are: " + sb.toString()); + } + } + + validateOptionalPositiveIntProperty(config, ConsumerConfigConstants.SHARD_GETRECORDS_MAX, + "Invalid value given for maximum records per getRecords shard operation. Must be a valid non-negative integer value."); + + validateOptionalPositiveIntProperty(config, ConsumerConfigConstants.SHARD_GETRECORDS_RETRIES, + "Invalid value given for maximum retry attempts for getRecords shard operation. Must be a valid non-negative integer value."); + + validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_BASE, + "Invalid value given for get records operation base backoff milliseconds. Must be a valid non-negative long value."); + + validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_MAX, + "Invalid value given for get records operation max backoff milliseconds. Must be a valid non-negative long value."); + + validateOptionalPositiveDoubleProperty(config, ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_EXPONENTIAL_CONSTANT, + "Invalid value given for get records operation backoff exponential constant. Must be a valid non-negative double value."); + + validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.SHARD_GETRECORDS_INTERVAL_MILLIS, + "Invalid value given for getRecords sleep interval in milliseconds. Must be a valid non-negative long value."); + + validateOptionalPositiveIntProperty(config, ConsumerConfigConstants.SHARD_GETITERATOR_RETRIES, + "Invalid value given for maximum retry attempts for getShardIterator shard operation. Must be a valid non-negative integer value."); + + validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_BASE, + "Invalid value given for get shard iterator operation base backoff milliseconds. Must be a valid non-negative long value."); + + validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_MAX, + "Invalid value given for get shard iterator operation max backoff milliseconds. Must be a valid non-negative long value."); + + validateOptionalPositiveDoubleProperty(config, ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_EXPONENTIAL_CONSTANT, + "Invalid value given for get shard iterator operation backoff exponential constant. Must be a valid non-negative double value."); + + validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.SHARD_DISCOVERY_INTERVAL_MILLIS, + "Invalid value given for shard discovery sleep interval in milliseconds. Must be a valid non-negative long value."); + + validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_BASE, + "Invalid value given for describe stream operation base backoff milliseconds. Must be a valid non-negative long value."); + + validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_MAX, + "Invalid value given for describe stream operation max backoff milliseconds. Must be a valid non-negative long value."); + + validateOptionalPositiveDoubleProperty(config, ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_EXPONENTIAL_CONSTANT, + "Invalid value given for describe stream operation backoff exponential constant. Must be a valid non-negative double value."); + + if (config.containsKey(ConsumerConfigConstants.SHARD_GETRECORDS_INTERVAL_MILLIS)) { + checkArgument( + Long.parseLong(config.getProperty(ConsumerConfigConstants.SHARD_GETRECORDS_INTERVAL_MILLIS)) + < ConsumerConfigConstants.MAX_SHARD_GETRECORDS_INTERVAL_MILLIS, + "Invalid value given for getRecords sleep interval in milliseconds. Must be lower than " + + ConsumerConfigConstants.MAX_SHARD_GETRECORDS_INTERVAL_MILLIS + " milliseconds." + ); + } + } + + /** + * Validate configuration properties for {@link FlinkKinesisProducer}. + */ + public static void validateProducerConfiguration(Properties config) { + checkNotNull(config, "config can not be null"); + + validateAwsConfiguration(config); + + validateOptionalPositiveLongProperty(config, ProducerConfigConstants.COLLECTION_MAX_COUNT, + "Invalid value given for maximum number of items to pack into a PutRecords request. Must be a valid non-negative long value."); + + validateOptionalPositiveLongProperty(config, ProducerConfigConstants.AGGREGATION_MAX_COUNT, + "Invalid value given for maximum number of items to pack into an aggregated record. Must be a valid non-negative long value."); + } + + /** + * Validate configuration properties related to Amazon AWS service + */ + public static void validateAwsConfiguration(Properties config) { + if (!config.containsKey(AWSConfigConstants.AWS_CREDENTIALS_PROVIDER)) { + // if the credential provider type is not specified, it will default to BASIC later on, + // so the Access Key ID and Secret Key must be given + if (!config.containsKey(AWSConfigConstants.AWS_ACCESS_KEY_ID) + || !config.containsKey(AWSConfigConstants.AWS_SECRET_ACCESS_KEY)) { + throw new IllegalArgumentException("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."); + } + } else { + String credentialsProviderType = config.getProperty(AWSConfigConstants.AWS_CREDENTIALS_PROVIDER); + + // value specified for AWSConfigConstants.AWS_CREDENTIALS_PROVIDER needs to be recognizable + CredentialProvider providerType; + try { + providerType = CredentialProvider.valueOf(credentialsProviderType); + } catch (IllegalArgumentException e) { + StringBuilder sb = new StringBuilder(); + for (CredentialProvider type : CredentialProvider.values()) { + sb.append(type.toString()).append(", "); + } + throw new IllegalArgumentException("Invalid AWS Credential Provider Type set in config. Valid values are: " + sb.toString()); + } + + // if BASIC type is used, also check that the Access Key ID and Secret Key is supplied + if (providerType == CredentialProvider.BASIC) { + if (!config.containsKey(AWSConfigConstants.AWS_ACCESS_KEY_ID) + || !config.containsKey(AWSConfigConstants.AWS_SECRET_ACCESS_KEY)) { + throw new IllegalArgumentException("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."); + } + } + } + + if (!config.containsKey(AWSConfigConstants.AWS_REGION)) { + throw new IllegalArgumentException("The AWS region ('" + AWSConfigConstants.AWS_REGION + "') must be set in the config."); + } else { + // specified AWS Region name must be recognizable + if (!AWSUtil.isValidRegion(config.getProperty(AWSConfigConstants.AWS_REGION))) { + StringBuilder sb = new StringBuilder(); + for (Regions region : Regions.values()) { + sb.append(region.getName()).append(", "); + } + throw new IllegalArgumentException("Invalid AWS region set in config. Valid values are: " + sb.toString()); + } + } + } + + private static void validateOptionalPositiveLongProperty(Properties config, String key, String message) { + if (config.containsKey(key)) { + try { + long value = Long.parseLong(config.getProperty(key)); + if (value < 0) { + throw new NumberFormatException(); + } + } catch (NumberFormatException e) { + throw new IllegalArgumentException(message); + } + } + } + + private static void validateOptionalPositiveIntProperty(Properties config, String key, String message) { + if (config.containsKey(key)) { + try { + int value = Integer.parseInt(config.getProperty(key)); + if (value < 0) { + throw new NumberFormatException(); + } + } catch (NumberFormatException e) { + throw new IllegalArgumentException(message); + } + } + } + + private static void validateOptionalPositiveDoubleProperty(Properties config, String key, String message) { + if (config.containsKey(key)) { + try { + double value = Double.parseDouble(config.getProperty(key)); + if (value < 0) { + throw new NumberFormatException(); + } + } catch (NumberFormatException e) { + throw new IllegalArgumentException(message); + } + } + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-connectors/flink-connector-kinesis/src/main/resources/log4j.properties ---------------------------------------------------------------------- diff --git a/flink-connectors/flink-connector-kinesis/src/main/resources/log4j.properties b/flink-connectors/flink-connector-kinesis/src/main/resources/log4j.properties new file mode 100644 index 0000000..773f932 --- /dev/null +++ b/flink-connectors/flink-connector-kinesis/src/main/resources/log4j.properties @@ -0,0 +1,27 @@ +################################################################################ +# 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. +################################################################################ + +log4j.rootLogger=INFO, testlogger + +log4j.appender.testlogger=org.apache.log4j.ConsoleAppender +log4j.appender.testlogger.target = System.err +log4j.appender.testlogger.layout=org.apache.log4j.PatternLayout +log4j.appender.testlogger.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n + +# suppress the irrelevant (wrong) warnings from the netty channel handler +log4j.logger.org.jboss.netty.channel.DefaultChannelPipeline=ERROR, testlogger
