Support SSL authentication with Kafka in routine load job by morningman · Pull Request #1235 · apache/doris

morningman · 2019-06-03T05:09:31Z

Implement a Small File Manager which allows user to upload small files, save them in Doris and use them later. Details can be found in docs/documentation/cn/administrator-guide/small_file_mgr.md and docs/help/Contents/Administration/small_files.md.
Support SSL authentication with Kafka in routine load job. Details can be found in docs/documentation/cn/administrator-guide/load-data/routine-load-manual.md and docs/help/Contents/Data Manipulation/routine_load.md
Remove Kafka Java client. Now we only use librdkafka on BE to connect Kafka brokers.
Fix bugs that invalid routine load task configuration may cause too many aborted transactions.
Modify the get_log_file restful api on FE. Details can be found in docs/documentation/cn/administrator-guide/http-actions/fe_get_log_file.md.
Optimize the error messages of CANCEL LOAD operation.

…ad-manual.md Co-Authored-By: kangpinghuang <40422952+kangpinghuang@users.noreply.github.com>

…log_file.md Co-Authored-By: kangkaisen <kangkaisen@apache.org>

kangkaisen

Save the small file content to FE metadata is necessary? could we save the small file content in FE disk or BE disk?

ghost · 2019-06-03T07:15:05Z

@@ -0,0 +1,71 @@
+package org.apache.doris.common.util;


licence header

ghost · 2019-06-03T07:18:06Z

@@ -0,0 +1,114 @@
+package org.apache.doris.common.util;


license header

Co-Authored-By: kangkaisen <kangkaisen@apache.org>

ghost · 2019-06-04T02:37:55Z

@kangkaisen
Save the small file content to FE metadata is necessary? could we save the small file content in FE disk or BE disk?

I've been considered this problem, here are my thoughts:

Limit the size and num of files so that FE memory is not a problem.
If files be saved on disk, we may need another file manager to manage files on disks. If files are damaged or missing, there is no where to download them again. And I just want this light small file manager to be a certain level of 'Stateless'.
BE also has a file cache, all downloaded files are saved on BE disks, and if file is missing, BE will downloaded file from FE again.
I also add an option which allow user to just save file url in SmallFileMgr(You can see it in CreateFileStmt.java, but because it is not fully implemented yet, I currently forbid this feature). With this option, file in Doris will be represented as a URL. And the validation of URL is guaranteed by user themselves.

ghost · 2019-06-04T03:11:52Z

@kangkaisen
Save the small file content to FE metadata is necessary? could we save the small file content in FE disk or BE disk?
I've been considered this problem, here are my thoughts:
Limit the size and num of files so that FE memory is not a problem.
If files be saved on disk, we may need another file manager to manage files on disks. If files are damaged or missing, there is no where to download them again. And I just want this light small file manager to be a certain level of 'Stateless'.
BE also has a file cache, all downloaded files are saved on BE disks, and if file is missing, BE will downloaded file from FE again.
I also add an option which allow user to just save file url in SmallFileMgr(You can see it in CreateFileStmt.java, but because it is not fully implemented yet, I currently forbid this feature). With this option, file in Doris will be represented as a URL. And the validation of URL is guaranteed by user themselves.

Ok. I See. Thank you.

ghost · 2019-06-04T15:05:01Z


 Status.TStatus submit_routine_load_task(1:list<TRoutineLoadTask> tasks);
+ // This is used for getting some information via Backend
+ TProxyResult get_info(1:TProxyRequest request);


Actually, new interface should use Brpc, and old interface should be changed to Brpc.
Because Brpc is better for example it support async, connection reused

OK, I will change it

ghost · 2019-06-04T15:16:53Z

+ // check md5sum if necessary
+ String checksum = Hex.encodeHexString(digest.digest());
+ if (!Strings.isNullOrEmpty(md5sum)) {
+ if (!checksum.equals(md5sum)) {


ignore case?

ghost · 2019-06-04T15:18:18Z

+ throw new DdlException("Failed to check md5 of file: " + file.getName());
+ }
+
+ return md5sum.equals(expectedMd5);


ignore case?

ghost · 2019-06-04T15:20:50Z

+ SmallFileMgr fileMgr = Catalog.getCurrentCatalog().getSmallFileMgr();
+ String filePath;
+ try {
+ filePath = fileMgr.saveToFile(fileId);


why do we need to save it in a file? I think we can get in memory and send the result.

ghost · 2019-06-04T15:22:52Z

+ if (entry.getValue().startsWith("FILE:")) {
+ String file = entry.getValue().substring(entry.getValue().indexOf(":") + 1);
+ // check and save file to disk
+ smallFileMgr.saveToFile(dbId, KAFKA_FILE_CATALOG, file);


I don't know why we should save it to file?

Co-Authored-By: ZHAO Chun <buaa.zhaoc@gmail.com>

ghost · 2019-06-05T01:56:13Z

temporary close

ghost · 2019-06-05T02:41:23Z

 } else {
 errMsg = "process loading finished fail";
-processCancelled(job, cancelType, errMsg);
+ processCancelled(job, cancelType, errMsg, failedMsg);


please keep neat

ghost · 2019-06-05T02:41:33Z

 break;
 case CANCELLED:
-processCancelled(job, cancelType, errMsg);
+ processCancelled(job, cancelType, errMsg, failedMsg);


please keep neat

ghost · 2019-06-05T03:33:40Z

+ CREATE FILE "ca.pem"
+ PROPERTIES
+ (
+ "url" = "https://test.bj.bcebos.com/kafka-key/ca.pem",


http or https？

this is just an example. both https and http is OK

ghost · 2019-06-06T02:56:00Z

-
- // this is the kafka consumer which is used to fetch the number of partitions
- private KafkaConsumer<String, String> consumer;
+ private Map<String, String> convertedCustomProperties = Maps.newHashMap();


why don't you merge convertedCustomProperties into customKafkaProperties

convertedCustomProperties is just a temporary data struct, and it should be re-created if file changed

ghost · 2019-06-06T03:04:42Z

- return result;
+ private List<Integer> getAllKafkaPartitions() throws UserException {
+ convertCustomProperties();
+ return KafkaUtil.getAllKafkaPartitions(brokerList, topic, convertedCustomProperties);


If the convertedCustomProperties is incorrect, the job will not be cancelled? The connection timeout and certification failed need to be distinguish.

Yes, if convertedCustomProperties is incorrect, the job will be paused.
And it is not easy to distinguish timeout and other failure. for example, if ssl authentication failed, kafka client on BE will retry again and again , while in FE, it looks like timeout.

Of cause, you remind me that convertedCustomProperties should be re-created when routine load job transfer from PAUSE to NEED_SCHEDULE, because some files may be re-created anytime.

ghost · 2019-06-06T03:14:37Z

+
+/*
+ * Author: Chenmingyu
+ * Date: May 29, 2019


em... Is author necessary ?

No, this is just the IDE's auto adding

ghost · 2019-06-06T03:24:17Z

 private static final Logger LOG = LogManager.getLogger(KafkaRoutineLoadJob.class);

-private static final int FETCH_PARTITIONS_TIMEOUT_SECOND = 5;
+public static final String KAFKA_FILE_CATALOG = "kafka";


The catalog is a static param of job while the catalog of file is created by user.

Yes, in my design , catalog named 'kafka' is a reserved keyword, which is exactly used by kafka client.

morningman added 22 commits May 30, 2019 22:01

add small file manager
117e405

fix bug1
7098512

fix bug2
cc989fd

fix bug3
3b2ed78

fix get log action
5315f3a

add small file write
e98aa5d

fix download bug
66e6e3f

fix too many abort tasks
8e58645

add FILE:
5bb8dec

add GetSmallFileAction
96e82f5

register getSmallFileAction
efb992a

no password
870db7c

add heartbeat master http port
dc6271a

replay create file
33a00f9

modify BE
822d6e9

run ok
d5d6f0f

FE get partitions from BE
816acb6

first ok
e034cac

FE works
961c4c6

modify cancel load msg
20f7e28

Add BE ut
63f3f23

add get_log_file doc
4aafe7b

kangkaisen reviewed Jun 3, 2019
View reviewed changes

Comment threaddocs/documentation/cn/administrator-guide/http-actions/fe_get_log_file.md Outdated

kangpinghuang requested changes Jun 3, 2019
View reviewed changes

Comment threaddocs/documentation/cn/administrator-guide/load-data/routine-load-manual.md Outdated

morningmanand others added 2 commits June 3, 2019 14:31

Update docs/documentation/cn/administrator-guide/load-data/routine-lo…
5d72da0
…ad-manual.md Co-Authored-By: kangpinghuang <40422952+kangpinghuang@users.noreply.github.com>

Update docs/documentation/cn/administrator-guide/http-actions/fe_get_…
83261ed
…log_file.md Co-Authored-By: kangkaisen <kangkaisen@apache.org>

kangkaisen reviewed Jun 3, 2019
View reviewed changes

Comment threaddocs/help/Contents/Data Definition/ddl_stmt.md Outdated
Comment threadfe/src/main/java/org/apache/doris/analysis/CreateFileStmt.java Outdated

imay requested changes Jun 3, 2019
View reviewed changes

Mingyu Chenand others added 2 commits June 4, 2019 09:10

Update fe/src/main/java/org/apache/doris/analysis/CreateFileStmt.java
5eeef55
Co-Authored-By: kangkaisen <kangkaisen@apache.org>

Update docs/help/Contents/Data Definition/ddl_stmt.md
d1caa69
Co-Authored-By: kangkaisen <kangkaisen@apache.org>

morningman added 3 commits June 4, 2019 09:54

add url support for create file stmt
903d0e1

fix bugs
d187499

fix bugs2
9fea318

imay requested changes Jun 4, 2019
View reviewed changes

Mingyu Chenand others added 2 commits June 5, 2019 09:08

Update be/src/runtime/routine_load/data_consumer.cpp
d423a05
Co-Authored-By: ZHAO Chun <buaa.zhaoc@gmail.com>

Update be/src/runtime/routine_load/data_consumer.cpp
be2ce1d
Co-Authored-By: ZHAO Chun <buaa.zhaoc@gmail.com>

morningman closed this Jun 5, 2019

morningman added 2 commits June 5, 2019 09:56

modify by review 1
2997f5f

change the way to get small file
ab70d3f

EmmyMiao87 reviewed Jun 6, 2019
View reviewed changes

modify to way of using proto
96681e5

morningman reopened this Jun 6, 2019

change routine load thread pool limit to 80
d3911b6

imay reviewed Jun 6, 2019
View reviewed changes

Comment threadgensrc/proto/internal_service.proto Outdated

remove saveTofile() in smallFileMgr
ec563c9

imay approved these changes Jun 7, 2019
View reviewed changes

imay merged commit ff0dd0d into apache:masterJun 7, 2019

Uh oh!

Conversation

morningman commented Jun 3, 2019• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

kangkaisen left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

ghostJun 3, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 4, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 3, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 4, 2019

Choose a reason for hiding this comment

Uh oh!

ghost commented Jun 4, 2019• edited by morningman Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ghost commented Jun 4, 2019

Uh oh!

Uh oh!

Uh oh!

ghostJun 4, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 5, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 4, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 5, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 4, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 5, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 4, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 4, 2019

Choose a reason for hiding this comment

Uh oh!

ghost commented Jun 5, 2019

Uh oh!

ghostJun 5, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 6, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 5, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 6, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 5, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 6, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 6, 2019

Choose a reason for hiding this comment

Uh oh!

ghostJun 6, 2019

Choose a reason for hiding this comment

morningman commented Jun 3, 2019•
edited
Loading

ghost commented Jun 4, 2019•
edited by morningman
Loading