EmmyMiao87 commented on a change in pull request #754: Add persist operations
for routine load job
URL: https://github.com/apache/incubator-doris/pull/754#discussion_r265537284
##########
File path:
fe/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java
##########
@@ -313,42 +309,52 @@ private void checkKafkaCustomProperties() throws
AnalysisException {
+ " not match pattern " +
ENDPOINT_REGEX);
}
}
+
// check topic
- kafkaTopic = customProperties.get(KAFKA_TOPIC_PROPERTY);
+ kafkaTopic =
Strings.nullToEmpty(dataSourceProperties.get(KAFKA_TOPIC_PROPERTY)).replaceAll("
", "");
if (Strings.isNullOrEmpty(kafkaTopic)) {
- throw new AnalysisException(KAFKA_TOPIC_PROPERTY + " is required
property");
+ throw new AnalysisException(KAFKA_TOPIC_PROPERTY + " is a required
property");
}
+
// check partitions
- final String kafkaPartitionsString =
customProperties.get(KAFKA_PARTITIONS_PROPERTY);
+ final String kafkaPartitionsString =
dataSourceProperties.get(KAFKA_PARTITIONS_PROPERTY);
if (kafkaPartitionsString != null) {
- kafkaPartitions = new ArrayList<>();
- if (kafkaPartitionsString.equals(EMPTY_STRING)) {
+ kafkaPartitionsString.replaceAll(" ", "");
+ if (kafkaPartitionsString.isEmpty()) {
throw new AnalysisException(KAFKA_PARTITIONS_PROPERTY + "
could not be a empty string");
}
String[] kafkaPartionsStringList =
kafkaPartitionsString.split(",");
for (String s : kafkaPartionsStringList) {
try {
- kafkaPartitions.add(getIntegerValueFromString(s,
KAFKA_PARTITIONS_PROPERTY));
+
kafkaPartitionOffsets.add(Pair.create(getIntegerValueFromString(s,
KAFKA_PARTITIONS_PROPERTY), 0L));
} catch (AnalysisException e) {
throw new AnalysisException(KAFKA_PARTITIONS_PROPERTY
+ " must be a number
string with comma-separated");
}
}
}
- // check offsets
- // Todo(ml)
- final String kafkaOffsetsString =
customProperties.get(KAFKA_OFFSETS_PROPERTY);
+
+ // check offset
+ final String kafkaOffsetsString =
dataSourceProperties.get(KAFKA_OFFSETS_PROPERTY);
if (kafkaOffsetsString != null) {
- kafkaOffsets = new ArrayList<>();
- String[] kafkaOffsetsStringList =
customProperties.get(KAFKA_OFFSETS_PROPERTY).split(",");
- for (String s : kafkaOffsetsStringList) {
- kafkaOffsets.add(Long.valueOf(s));
+ kafkaOffsetsString.replaceAll(" ", "");
+ if (kafkaOffsetsString.isEmpty()) {
+ throw new AnalysisException(KAFKA_OFFSETS_PROPERTY + " could
not be a empty string");
+ }
+ String[] kafkaOffsetsStringList = kafkaOffsetsString.split(",");
+ if (kafkaOffsetsStringList.length != kafkaPartitionOffsets.size())
{
Review comment:
if some partition has offset while others hasn't ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
