You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2019/02/22 11:21:04 UTC
[flink] 02/04: [FLINK-8354] Fix formatting in FlinkKafkaConsumerBase
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit fa403a9998328a81ec6c3c785e842526f6abba0a
Author: Alexey Trenikhin <al...@gmail.com>
AuthorDate: Wed Feb 20 14:03:26 2019 +0100
[FLINK-8354] Fix formatting in FlinkKafkaConsumerBase
---
.../connectors/kafka/FlinkKafkaConsumerBase.java | 16 ++++++++--------
1 file changed, 8 insertions(+), 8 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
index 196b5c0..bf8f29e 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
@@ -497,7 +497,7 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
// restored partitions that should not be subscribed by this subtask
if (KafkaTopicPartitionAssigner.assign(
restoredStateEntry.getKey(), getRuntimeContext().getNumberOfParallelSubtasks())
- == getRuntimeContext().getIndexOfThisSubtask()){
+ == getRuntimeContext().getIndexOfThisSubtask()){
subscribedPartitionsToStartOffsets.put(restoredStateEntry.getKey(), restoredStateEntry.getValue());
}
} else {
@@ -545,16 +545,16 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
}
for (Map.Entry<KafkaTopicPartition, Long> partitionToOffset
- : fetchOffsetsWithTimestamp(allPartitions, startupOffsetsTimestamp).entrySet()) {
+ : fetchOffsetsWithTimestamp(allPartitions, startupOffsetsTimestamp).entrySet()) {
subscribedPartitionsToStartOffsets.put(
partitionToOffset.getKey(),
(partitionToOffset.getValue() == null)
- // if an offset cannot be retrieved for a partition with the given timestamp,
- // we default to using the latest offset for the partition
- ? KafkaTopicPartitionStateSentinel.LATEST_OFFSET
- // since the specified offsets represent the next record to read, we subtract
- // it by one so that the initial state of the consumer will be correct
- : partitionToOffset.getValue() - 1);
+ // if an offset cannot be retrieved for a partition with the given timestamp,
+ // we default to using the latest offset for the partition
+ ? KafkaTopicPartitionStateSentinel.LATEST_OFFSET
+ // since the specified offsets represent the next record to read, we subtract
+ // it by one so that the initial state of the consumer will be correct
+ : partitionToOffset.getValue() - 1);
}
break;