KevinyhZou commented on code in PR #13409:
URL: https://github.com/apache/gluten/pull/13409#discussion_r4464390825
##########
gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecExchange.java:
##########
@@ -153,8 +153,12 @@ protected Transformation<RowData> translateToPlanInternal(
LogicalTypeConverter.toVLType(getOutputType());
String id = PlanNodeIdGenerator.newId();
List<Integer> keyIndexes =
Arrays.stream(keys).boxed().collect(Collectors.toList());
+ // Keyed shuffle routes by key group (the same value the state
+ // backend derives key groups with), so each subtask receives
+ // exactly the key groups its state owns.
PartitionFunctionSpec partitionFunctionSpec =
- new HashPartitionFunctionSpec(glutenInputType, keyIndexes);
+ new KeyGroupPartitionFunctionSpec(
+ glutenInputType, keyIndexes,
DEFAULT_LOWER_BOUND_MAX_PARALLELISM);
Review Comment:
Could we pass the exchange/operator actual max parallelism here instead of
fixing it to 128? Flink may configure the downstream partitioner and keyed
state with a different max parallelism (for example, an explicitly configured
value). In that case the native spec computes key groups modulo 128 while state
ownership uses the actual value, so records can be sent to a subtask that does
not own their state. Please also cover a non-128 max-parallelism case in a
routing test.
##########
gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java:
##########
@@ -419,15 +454,30 @@ private boolean checkJobRunningStatus(TableResult result,
long timeoutMs)
if (jobClient.getJobStatus().complete(JobStatus.RUNNING)) {
jobClient.cancel();
return true;
- } else {
- Thread.sleep(1000);
}
+ if (jobClient.getJobStatus().complete(JobStatus.FAILED)) {
+ LOG.warn("Job failed before reaching RUNNING state.");
+ return jobFailedOnUnsupportedFeature(jobClient);
+ }
+ Thread.sleep(1000);
}
LOG.warn("Job not running in " + timeoutMs + " millseconds.");
jobClient.cancel();
return false;
}
+ private boolean jobFailedOnUnsupportedFeature(JobClient jobClient) throws
InterruptedException {
+ try {
+ jobClient.getJobExecutionResult().get();
+ } catch (ExecutionException e) {
+ if (isCausedByUnsupportedFeature(e)) {
+ LOG.warn("Job failed on an unsupported stream feature; tolerating the
failure.", e);
+ return true;
Review Comment:
Returning true here makes checkJobRunningStatus() report success even though
the job reached FAILED before RUNNING. The Kafka-source tests assert that
return value as proof that the job started, so they can now pass without
exercising a running job. Could we represent an unsupported-feature skip
separately (for example, with a test assumption) and reserve true for a job
that actually reached RUNNING?
--
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]
