add kafka_default_offsets when no partiotion specify .support read kafka partition from start #1642

wkhappy1 · 2019-08-14T09:28:18Z

add kafka_default_offsets when no partiotion specify
value OFFSET_BEGINNING,OFFSET_END

imay · 2019-08-14T11:51:56Z

fe/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java

@@ -94,7 +94,7 @@
    // optional
    public static final String KAFKA_PARTITIONS_PROPERTY = "kafka_partitions";
    public static final String KAFKA_OFFSETS_PROPERTY = "kafka_offsets";
-
+    public static final String KAFKA_DEFAULT_OFFSETS="kafka_default_offsets";


Suggested change

public static final String KAFKA_DEFAULT_OFFSETS="kafka_default_offsets";

public static final String KAFKA_DEFAULT_OFFSETS = "kafka_default_offsets";

imay · 2019-08-14T11:52:07Z

fe/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java

@@ -136,6 +137,8 @@
    private String kafkaTopic;
    // pair<partition id, offset>
    private List<Pair<Integer, Long>> kafkaPartitionOffsets = Lists.newArrayList();
+    // kafka default offset
+    private String kafkaDefaultOffset="";


Suggested change

private String kafkaDefaultOffset="";

private String kafkaDefaultOffset = "";

imay · 2019-08-14T11:52:30Z

fe/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java

@@ -401,6 +407,11 @@ private void checkKafkaProperties() throws AnalysisException {
                }
            }
        }
+        // check default offset
+        final String kafkaODefaultffsetsString = dataSourceProperties.get(KAFKA_DEFAULT_OFFSETS);


Suggested change

final String kafkaODefaultffsetsString = dataSourceProperties.get(KAFKA_DEFAULT_OFFSETS);

final String kafkaDefaultOffsetsString = dataSourceProperties.get(KAFKA_DEFAULT_OFFSETS);

imay · 2019-08-14T11:53:13Z

fe/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java

@@ -368,11 +370,21 @@ private void updateNewPartitionProgress() {
        for (Integer kafkaPartition : currentKafkaPartitions) {
            if (!((KafkaProgress) progress).containsPartition(kafkaPartition)) {
                // if offset is not assigned, start from OFFSET_END
-                ((KafkaProgress) progress).addPartitionOffset(Pair.create(kafkaPartition, KafkaProgress.OFFSET_END_VAL));
+                long beginOffSet = KafkaProgress.OFFSET_END_VAL;
+                if(!kafkaDefaultOffSet.isEmpty()) {


Suggested change

if(!kafkaDefaultOffSet.isEmpty()) {

if (!kafkaDefaultOffSet.isEmpty()) {

imay · 2019-08-14T11:53:57Z

fe/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java

-                ((KafkaProgress) progress).addPartitionOffset(Pair.create(kafkaPartition, KafkaProgress.OFFSET_END_VAL));
+                long beginOffSet = KafkaProgress.OFFSET_END_VAL;
+                if(!kafkaDefaultOffSet.isEmpty()) {
+                    if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_BEGINNING)) {


Suggested change

if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_BEGINNING)) {

if (kafkaDefaultOffSet.equalsIgnoreCase(KafkaProgress.OFFSET_BEGINNING)) {

imay · 2019-08-14T11:54:13Z

fe/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java

+                if(!kafkaDefaultOffSet.isEmpty()) {
+                    if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_BEGINNING)) {
+                        beginOffSet = KafkaProgress.OFFSET_BEGINNING_VAL;
+                    }else if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_END)) {


Suggested change

}else if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_END)) {

} else if (kafkaDefaultOffSet.equalsIgnoreCase(KafkaProgress.OFFSET_END)) {

morningman · 2019-08-14T11:56:57Z

docs/documentation/cn/sql-reference/sql-statements/Data Manipulation/ROUTINE LOAD.md

-            4. property
+            4. kafka_default_offsets 
+               指定kafka partition的默认起始offset
+               如果没有指定kafka_partitions/kafka_offsets,默认消费所有分区,此时可以指定kafka_default_offsets指定起始 offset。


Suggested change

如果没有指定kafka_partitions/kafka_offsets,默认消费所有分区,此时可以指定kafka_default_offsets指定起始 offset。

如果没有指定kafka_partitions/kafka_offsets,默认消费所有分区,此时可以指定kafka_default_offsets指定起始 offset。默认为 OFFSET_END，即从末尾开始订阅。

morningman · 2019-08-14T11:57:04Z

fe/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java

@@ -94,7 +94,7 @@
    // optional
    public static final String KAFKA_PARTITIONS_PROPERTY = "kafka_partitions";
    public static final String KAFKA_OFFSETS_PROPERTY = "kafka_offsets";
-
+    public static final String KAFKA_DEFAULT_OFFSETS="kafka_default_offsets";


Suggested change

public static final String KAFKA_DEFAULT_OFFSETS="kafka_default_offsets";

public static final String KAFKA_DEFAULT_OFFSETS = "kafka_default_offsets";

morningman · 2019-08-14T11:57:11Z

docs/documentation/cn/sql-reference/sql-statements/Data Manipulation/ROUTINE LOAD.md

+                  2) OFFSET_END: 从末尾开始订阅。
+               示例：
+
+                    "kafka_default_offsets"="OFFSET_BEGINNING"


Suggested change

"kafka_default_offsets"="OFFSET_BEGINNING"

"kafka_default_offsets" = "OFFSET_BEGINNING"

morningman · 2019-08-14T11:57:26Z

fe/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java

@@ -136,6 +137,8 @@
    private String kafkaTopic;
    // pair<partition id, offset>
    private List<Pair<Integer, Long>> kafkaPartitionOffsets = Lists.newArrayList();
+    // kafka default offset
+    private String kafkaDefaultOffset="";


Suggested change

private String kafkaDefaultOffset="";

private String kafkaDefaultOffset = "";

morningman · 2019-08-14T11:58:31Z

fe/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java

@@ -401,6 +407,11 @@ private void checkKafkaProperties() throws AnalysisException {
                }
            }
        }
+        // check default offset
+        final String kafkaODefaultffsetsString = dataSourceProperties.get(KAFKA_DEFAULT_OFFSETS);


Suggested change

final String kafkaODefaultffsetsString = dataSourceProperties.get(KAFKA_DEFAULT_OFFSETS);

final String kafkaDefaultOffsetsString = dataSourceProperties.get(KAFKA_DEFAULT_OFFSETS);

morningman · 2019-08-14T12:00:53Z

fe/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java

+                if(!kafkaDefaultOffSet.isEmpty()) {
+                    if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_BEGINNING)) {
+                        beginOffSet = KafkaProgress.OFFSET_BEGINNING_VAL;
+                    }else if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_END)) {


Suggested change

}else if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_END)) {

} else if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_END)) {

morningman · 2019-08-14T12:01:04Z

fe/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java

+                        beginOffSet = KafkaProgress.OFFSET_BEGINNING_VAL;
+                    }else if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_END)) {
+                        beginOffSet = KafkaProgress.OFFSET_END_VAL;
+                    }else {


Suggested change

}else {

} else {

morningman · 2019-08-14T12:04:57Z

fe/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java

@@ -368,11 +370,21 @@ private void updateNewPartitionProgress() {
        for (Integer kafkaPartition : currentKafkaPartitions) {
            if (!((KafkaProgress) progress).containsPartition(kafkaPartition)) {
                // if offset is not assigned, start from OFFSET_END
-                ((KafkaProgress) progress).addPartitionOffset(Pair.create(kafkaPartition, KafkaProgress.OFFSET_END_VAL));
+                long beginOffSet = KafkaProgress.OFFSET_END_VAL;
+                if(!kafkaDefaultOffSet.isEmpty()) {


Suggested change

if(!kafkaDefaultOffSet.isEmpty()) {

if (!kafkaDefaultOffSet.isEmpty()) {

morningman · 2019-08-14T12:05:05Z

fe/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java

-                ((KafkaProgress) progress).addPartitionOffset(Pair.create(kafkaPartition, KafkaProgress.OFFSET_END_VAL));
+                long beginOffSet = KafkaProgress.OFFSET_END_VAL;
+                if(!kafkaDefaultOffSet.isEmpty()) {
+                    if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_BEGINNING)) {


Suggested change

if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_BEGINNING)) {

if (kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_BEGINNING)) {

imay

LGTM

fe/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java

…eLoadJob.java Co-Authored-By: Mingyu Chen <morningman@163.com>

morningman

LGTM

add kafka_default_offsets when no partiotion specify

de50b0e

imay requested changes Aug 14, 2019

View reviewed changes

morningman requested changes Aug 14, 2019

View reviewed changes

wkhappy1 added 3 commits August 15, 2019 11:36

Update ROUTINE LOAD.md

bb2b156

Update CreateRoutineLoadStmt.java

ccdae9e

Update KafkaRoutineLoadJob.java

5caff6d

imay previously approved these changes Aug 15, 2019

View reviewed changes

morningman requested changes Aug 15, 2019

View reviewed changes

fe/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java Outdated Show resolved Hide resolved

Update fe/src/main/java/org/apache/doris/load/routineload/KafkaRoutin…

c3f70cc

…eLoadJob.java Co-Authored-By: Mingyu Chen <morningman@163.com>

wkhappy1 dismissed imay’s stale review via c3f70cc August 15, 2019 08:24

morningman closed this Aug 16, 2019

morningman reopened this Aug 16, 2019

morningman approved these changes Aug 16, 2019

View reviewed changes

morningman merged commit 1ed25ad into apache:master Aug 16, 2019

wkhappy1 deleted the my branch August 16, 2019 06:28

imay mentioned this pull request Sep 26, 2019

Release Notes 0.11.0 #1891

Closed

SWJTU-ZhangLei pushed a commit to SWJTU-ZhangLei/incubator-doris that referenced this pull request Jul 25, 2023

Fix accept_null_predicate mem leak (apache#1642)

762d9b6

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

add kafka_default_offsets when no partiotion specify .support read kafka partition from start #1642

add kafka_default_offsets when no partiotion specify .support read kafka partition from start #1642

wkhappy1 commented Aug 14, 2019

imay Aug 14, 2019

imay Aug 14, 2019

imay Aug 14, 2019

imay Aug 14, 2019

imay Aug 14, 2019

imay Aug 14, 2019

morningman Aug 14, 2019

morningman Aug 14, 2019

morningman Aug 14, 2019

morningman Aug 14, 2019

morningman Aug 14, 2019

morningman Aug 14, 2019

morningman Aug 14, 2019

morningman Aug 14, 2019

morningman Aug 14, 2019

imay left a comment

morningman left a comment

	public static final String KAFKA_DEFAULT_OFFSETS="kafka_default_offsets";
	public static final String KAFKA_DEFAULT_OFFSETS = "kafka_default_offsets";

	private String kafkaDefaultOffset="";
	private String kafkaDefaultOffset = "";

	final String kafkaODefaultffsetsString = dataSourceProperties.get(KAFKA_DEFAULT_OFFSETS);
	final String kafkaDefaultOffsetsString = dataSourceProperties.get(KAFKA_DEFAULT_OFFSETS);

	if(!kafkaDefaultOffSet.isEmpty()) {
	if (!kafkaDefaultOffSet.isEmpty()) {

	if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_BEGINNING)) {
	if (kafkaDefaultOffSet.equalsIgnoreCase(KafkaProgress.OFFSET_BEGINNING)) {

	}else if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_END)) {
	} else if (kafkaDefaultOffSet.equalsIgnoreCase(KafkaProgress.OFFSET_END)) {

	如果没有指定kafka_partitions/kafka_offsets,默认消费所有分区,此时可以指定kafka_default_offsets指定起始 offset。
	如果没有指定kafka_partitions/kafka_offsets,默认消费所有分区,此时可以指定kafka_default_offsets指定起始 offset。默认为 OFFSET_END，即从末尾开始订阅。

	"kafka_default_offsets"="OFFSET_BEGINNING"
	"kafka_default_offsets" = "OFFSET_BEGINNING"

	}else if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_END)) {
	} else if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_END)) {

	if(kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_BEGINNING)) {
	if (kafkaDefaultOffSet.equals(KafkaProgress.OFFSET_BEGINNING)) {

add kafka_default_offsets when no partiotion specify .support read kafka partition from start #1642

add kafka_default_offsets when no partiotion specify .support read kafka partition from start #1642

Conversation

wkhappy1 commented Aug 14, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

imay left a comment

Choose a reason for hiding this comment

morningman left a comment

Choose a reason for hiding this comment