Copilot commented on code in PR #679:
URL:
https://github.com/apache/doris-flink-connector/pull/679#discussion_r3654036234
##########
flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/datastream/DorisSourceFunction.java:
##########
@@ -91,7 +91,7 @@ public void run(SourceContext<List<?>> sourceContext) {
try (DorisValueReader valueReader =
new DorisValueReader(partitions, options, readOptions)) {
while (isRunning && valueReader.hasNext()) {
- List<?> next = valueReader.next();
+ List<?> next = valueReader.next().getFieldValues();
sourceContext.collect(next);
Review Comment:
valueReader.next().getFieldValues() returns an unmodifiable List (see
DorisSourceRecord), which changes the emitted runtime type compared to the
previous behavior (mutable list from the reader). Emitting a mutable copy here
keeps compatibility and avoids potential serialization / user-code assumptions
about mutability.
##########
flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/datastream/DorisSourceFunction.java:
##########
@@ -91,7 +91,7 @@ public void run(SourceContext<List<?>> sourceContext) {
try (DorisValueReader valueReader =
new DorisValueReader(partitions, options, readOptions)) {
while (isRunning && valueReader.hasNext()) {
- List<?> next = valueReader.next();
+ List<?> next = valueReader.next().getFieldValues();
sourceContext.collect(next);
Review Comment:
valueReader.next().getFieldValues() returns an unmodifiable List (see
DorisSourceRecord), which changes the emitted runtime type compared to the
previous behavior (mutable list from the reader). Emitting a mutable copy here
keeps compatibility and avoids potential serialization / user-code assumptions
about mutability.
##########
flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/rest/RestService.java:
##########
@@ -455,44 +632,123 @@ public static Schema getSchema(
@VisibleForTesting
public static JsonNode handleResponse(HttpUriRequest request, Logger
logger) {
try (CloseableHttpClient httpclient = HttpClients.createDefault()) {
- CloseableHttpResponse response = httpclient.execute(request);
- final int statusCode = response.getStatusLine().getStatusCode();
- final String reasonPhrase =
response.getStatusLine().getReasonPhrase();
- if (statusCode == 200 && response.getEntity() != null) {
- String responseEntity =
EntityUtils.toString(response.getEntity());
- return objectMapper.readTree(responseEntity);
- } else {
- throw new DorisRuntimeException(
- "Failed to parse response, status: "
- + statusCode
- + ", reason: "
- + reasonPhrase);
+ try (CloseableHttpResponse response = httpclient.execute(request))
{
+ final int statusCode =
response.getStatusLine().getStatusCode();
+ final String reasonPhrase =
response.getStatusLine().getReasonPhrase();
+ if (statusCode == 200 && response.getEntity() != null) {
+ String responseEntity =
EntityUtils.toString(response.getEntity());
+ return objectMapper.readTree(responseEntity);
+ } else {
+ throw new DorisRuntimeException(
+ "Failed to parse response, status: "
+ + statusCode
+ + ", reason: "
+ + reasonPhrase);
+ }
}
} catch (Exception e) {
logger.trace("request error,", e);
- throw new DorisRuntimeException("request error with " +
e.getMessage());
+ throw new DorisRuntimeException(
+ "request error for " + request.getURI() + ": " +
e.getMessage(), e);
}
}
- /** Try to get the ArrowFlightSqlPort port */
- public static Integer tryGetArrowFlightSqlPort(
- DorisOptions options, DorisReadOptions readOptions, Logger logger)
{
- if (readOptions.getFlightSqlPort() != null &&
readOptions.getFlightSqlPort() > 0) {
- return readOptions.getFlightSqlPort();
+ /** Executes a SQL statement through the Doris FE HTTP statement endpoint.
*/
+ public static JsonNode executeStatement(DorisOptions options, String
statement, Logger logger) {
+ try {
+ return executeStatementAtEndpoint(
+ options,
+ DorisReadOptions.defaults(),
+ randomEndpoint(options.getFenodes(), logger),
+ statement,
+ logger);
+ } catch (IllegalArgumentException e) {
+ throw new DorisRuntimeException("Failed to select a Doris FE
endpoint", e);
}
+ }
+
+ private static JsonNode executeStatementAtEndpoint(
+ DorisOptions options,
+ DorisReadOptions readOptions,
+ String endpoint,
+ String statement,
+ Logger logger) {
try {
Map<String, String> param = new HashMap<>();
- param.put("stmt", "show frontends");
- String requestUrl =
- String.format(STATEMENT_EXEC_API,
randomEndpoint(options.getFenodes(), logger));
+ param.put("stmt", statement);
+ String requestUrl = String.format(STATEMENT_EXEC_API, endpoint);
HttpPost httpPost = new HttpPost(requestUrl);
httpPost.setHeader(HttpHeaders.AUTHORIZATION, authHeader(options));
httpPost.setHeader(
HttpHeaders.CONTENT_TYPE,
String.format("application/json;charset=%s", "UTF-8"));
httpPost.setEntity(new
StringEntity(objectMapper.writeValueAsString(param), "UTF-8"));
+ httpPost.setConfig(createRequestConfig(readOptions));
JsonNode response = handleResponse(httpPost, logger);
+ if (response.has("code") && response.path("code").asInt() !=
REST_RESPONSE_CODE_OK) {
+ throw new DorisRuntimeException(
+ "Failed to execute Doris statement: " +
response.path("msg").asText());
+ }
+ return response;
+ } catch (DorisRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new DorisRuntimeException("Failed to execute Doris
statement", e);
+ }
+ }
+
+ @VisibleForTesting
+ static String parseScalarStatementResult(JsonNode rootNode) {
+ JsonNode dataNode = rootNode.path("data").path("data");
+ if (dataNode.size() != 1
+ || !dataNode.get(0).isArray()
+ || dataNode.get(0).size() != 1
+ || dataNode.get(0).get(0).isNull()) {
+ throw new DorisRuntimeException(
+ "Doris scalar statement must return exactly one row and
one column");
+ }
+ return dataNode.get(0).get(0).asText();
+ }
+
+ private static JsonNode getFrontends(
+ DorisOptions options, DorisReadOptions readOptions, String
endpoint, Logger logger) {
+ return executeStatementAtEndpoint(options, readOptions, endpoint,
"show frontends", logger);
+ }
+
+ @VisibleForTesting
+ static String getMasterFrontendEndpoint(JsonNode rootNode) {
+ JsonNode metaNode = rootNode.path("data").path("meta");
+ JsonNode dataNode = rootNode.path("data").path("data");
+ int hostIndex = findStatementColumnIndex(metaNode, "Host");
+ int httpPortIndex = findStatementColumnIndex(metaNode, "HttpPort");
+ int isMasterIndex = findStatementColumnIndex(metaNode, "IsMaster");
+
+ for (JsonNode row : dataNode) {
+ if (!row.isArray()
+ || row.size() <= Math.max(isMasterIndex,
Math.max(hostIndex, httpPortIndex))
+ || !Boolean.parseBoolean(row.get(isMasterIndex).asText()))
{
+ continue;
+ }
+ String host = row.get(hostIndex).asText();
+ int httpPort = row.get(httpPortIndex).asInt(-1);
+ if (StringUtils.isBlank(host) || httpPort <= 0 || httpPort >
65535) {
+ throw new DorisRuntimeException(
+ "Doris master FE has invalid Host or HttpPort in SHOW
FRONTENDS result");
+ }
+ return host + ":" + httpPort;
+ }
+ throw new DorisRuntimeException("No master FE found in SHOW FRONTENDS
result");
+ }
+
+ /** Try to get the ArrowFlightSqlPort port */
+ public static Integer tryGetArrowFlightSqlPort(
+ DorisOptions options, DorisReadOptions readOptions, Logger logger)
{
+ if (readOptions.getFlightSqlPort() != null &&
readOptions.getFlightSqlPort() > 0) {
+ return readOptions.getFlightSqlPort();
+ }
+ try {
+ JsonNode response = executeStatement(options, "show frontends",
logger);
logger.info("Get ArrowFlightSqlPort response is '{}'.", response);
return getArrowFlightSqlPort(response);
Review Comment:
tryGetArrowFlightSqlPort() currently executes SHOW FRONTENDS via
executeStatement(), which hard-codes DorisReadOptions.defaults() and therefore
ignores the caller-provided readOptions timeouts/retries. This can make Flight
SQL port discovery hang longer than configured or behave inconsistently with
other REST calls that respect readOptions.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
