https://flink.apache.org/ logo
Join Slack
Powered by
Hi! Is there any recommended way to be able to ba...
# troubleshooting
v
Hi! Is there any recommended way to be able to backport fixes to an older release and generate a 
tgz
from source similar to one of the official release downloads ? Context: We are trying to use Flink operator autoscaler (which is only recommended to be used with Flink 1.17) with our Beam pipelines. Beam doesn't have support yet for the latest Flink 1.17 (only has support till 1.16), hence we are trying to backport the recommended autoscaler related fixes to Flink 1.15 OR 1.16. What I tried: I tried following Building Flink from Source but the command gets stuck (without any errors) on my environment when building off the 1.15 release tag. Also, the 
mvn install
command doesn't generate files / directory structure similar to the release 
tgz
Any help on backporting these fixes and getting to a Flink java docker image would be appreciated!
Bumping this up. We are looking to build a Flink docker image from source, after backporting a few fixes from release version 1.17 to 1.16/1.15. Thanks!
a
Hi Well, cannot help you yet, but I am trying exactly same thing and I cant take 1.17 because beam is not there yet. Im in the process of compiling and current step is: [INFO] Building Flink : Connectors : SQL : Hive 3.1.2 1.16.2 [111/213] Can I ask what step you got stucked at? If I can get through this build maybe I can tell something, but thats still if. Building takes years and years.
hm... it finished compiling... I have now build-target directory which actually resembles this tgz directory... if not exactly same ( I think it is). I started with checkout on commit fb4d4ef57439cd7fde4fb5a4972733d13e69a226 then JAVA_HOME for openjdk-11, mvn 3.8.4 and it did work. I think at some point it used some node... $ node --version v14.21.2 And I used mvn package, not install.
v
(Sorry for the delayed response here, I fell sick and had been out since) Thanks @Alicja Celigowska for the details! My mvn install (and now package) seems to get stuck at:
Copy code
[INFO] Dependency-reduced POM written at: /Users/vaibhavgosain/Work/flink/flink/flink-test-utils-parent/flink-test-utils-junit/target/dependency-reduced-pom.xml
Similar to https://stackoverflow.com/questions/73561550/maven-build-gets-stuck and (possibly?) https://issues.apache.org/jira/browse/MSHADE-148 I downgraded my maven to 3.2.x as someone suggested but that doesnt seem to help as well
β€’ Thanks for the mvn package tip, good to know that that is the way to get to the tgz directory structure πŸ˜„ β€’ Interesting to see that it worked for you with mvn 3.8.4 , thanks for pointing out your versions
a
You were using 3.8.6 before? Or 3.8 branch? Maybe its a matter of commit ID?If you said before you used 1.15, then maybe lets try 1.16? I would suggest to use this maven and commit ID. Im bit surprised it does some test component even if -DskipTests option is used? Yes, tgz seems less of an issue... after building, I just made quick dockerfile: FROM flink:1.16.2-java11 ARG _BUILD_DIR_ ARG _FLINK_DIR_ WORKDIR /opt/flink/ COPY ${_FLINK_DIR_}/bin/ /opt/flink/bin/ COPY ${_FLINK_DIR_}/conf/ /opt/flink/conf/ COPY ${_FLINK_DIR_}/lib/ /opt/flink/lib/ COPY ${_FLINK_DIR_}/opt/ /opt/flink/opt/ COPY ${_FLINK_DIR_}/plugins/ /opt/flink/plugins/ RUN mkdir -p /opt/flink/plugins/azure-fs-hadoop COPY ${_FLINK_DIR_}/opt/flink-azure-fs-hadoop-1.16.2.jar /opt/flink/plugins/azure-fs-hadoop/ ------------------------- Its super dumb, but worked. I see my commit ID from cherry-pick in flink dashboard. I guess that you may also try "maven:3.8.4-openjdk-11" or something like that image to make "mvn package -DskipTests" and see if it passes there. I have my flink job running, it runs almost on idle yet, because simulation environment is still pending. Once we put some load, I want to verify autoscaling function. However, I did some quick test already: I did set parallelism to 2 for the job and I was wondering if it will downscale to 1. It actually did for all vertexes except kafka source. From flink dashboard I dont quite see however metrics as https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/datastream/kafka/... I started wondering, if this means autoscaler will keep skipping scaling this part of the job. While probably autiscaling later stage will be more important, at some point this one should too... maybe apache beam kafka source does not suport necessary ones? hmm... @Vaibhav Gosain If you use apache beam, then maybe you will run into same problem. It would be interesting if we could help each other?
Backpressured / Idle / BusyBackpressure 0% / 0% / N/A Thats what I have for sources part... I think this may be blocking downgrading parallelism to 1... KafkaIO maybe its not right choice, but then what is...
v
Thanks for more details I tried on 1.15.x as well as 1.16.x (also on the commit you provided) Tried maven 3.2 as well as 3.9 Note that I have been doing this on my Mac with the Apple Chip (in an env which has a lot of my company's dev deps installed) I am now trying to build on a fresh ubuntu EC2 instance 🀞
a
hm, thought about suggesting docker with maven 3.8 and java11, but what you said sounds like should do thins similarly. Let me know if it works, Im interested in how you will do actually πŸ˜… Do you actually use kafka as source? Or something else?
πŸ‘ 1
v
@Alicja Celigowska Yeah I am also thinking the same: As written in https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/docs/custom-resource/autoscaler/#autoscaler "Source scaling only supports modern sources which.." does the kafka source satisfy these conditions?
No we are not using kafka source. We use JDBC source for inputting from SQL, SqsIO for input from SQS, and a couple of custom written sources
a
Well, Im not exactly sure - I set parallelism 2, and its now set to 1 for everything but kafka source. Which makes me wonder if this is the case... I see, so you use different sources. But if I can ask later on, I wonder if parallelism for your sources will scale up or not - for those first tasks specifically. Just curious what will be experience
πŸ‘ 1
v
Yes I will keep you updated, thanks for the insights 🀝
a
Thanks πŸ™‚ I will be happy to read. Till next time
No busyTimeMsPerSecond metric available for - I get those kind of errors. Well, I do have this error even for kafka. I studied Source objects in apache beam bounded/unbounded, now studying flink code too. Checked bounded and it did not look well, but I wonder how native connectors use them...
Copy code
$ find . -name "*.java" | xargs grep TASK_BUSY_TIME 2>/dev/null
./flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTestBase.java:            Gauge<Double> busyTimeGauge = (Gauge<Double>) metrics.get(MetricNames.TASK_BUSY_TIME);
./flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandler.java:        return getMsPerSecondMetricAsRatio(metricStore, MetricNames.TASK_BUSY_TIME);
./flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/util/MutableIOMetrics.java:                    if (metrics.getMetric(MetricNames.ACC_TASK_BUSY_TIME) != null) {
./flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/util/MutableIOMetrics.java:                                        metrics.getMetric(MetricNames.ACC_TASK_BUSY_TIME));
./flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricNames.java:    public static final String TASK_BUSY_TIME = "busyTimeMs" + SUFFIX_RATE;
./flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricNames.java:    public static final String ACC_TASK_BUSY_TIME = "accumulateBusyTimeMs";
./flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskIOMetricGroup.java:        this.busyTimePerSecond = gauge(MetricNames.TASK_BUSY_TIME, this::getBusyTimePerSecond);
./flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskIOMetricGroup.java:                gauge(MetricNames.ACC_TASK_BUSY_TIME, this::getAccumulatedBusyTime);
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(task0, MetricNames.TASK_BUSY_TIME, "0"));
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(task1, MetricNames.TASK_BUSY_TIME, "900"));
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(task3, MetricNames.TASK_BUSY_TIME, "700"));
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(task0, MetricNames.TASK_BUSY_TIME, "0"));
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(speculativeTask0, MetricNames.TASK_BUSY_TIME, "800"));
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(task1, MetricNames.TASK_BUSY_TIME, "900"));
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(speculativeTask1, MetricNames.TASK_BUSY_TIME, "100"));
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(task3, MetricNames.TASK_BUSY_TIME, "700"));
Copy code
$ find . -name "*.java" | xargs grep TASK_BUSY_TIME 2>/dev/null
./flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTestBase.java:            Gauge<Double> busyTimeGauge = (Gauge<Double>) metrics.get(MetricNames.TASK_BUSY_TIME);
./flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandler.java:        return getMsPerSecondMetricAsRatio(metricStore, MetricNames.TASK_BUSY_TIME);
./flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/util/MutableIOMetrics.java:                    if (metrics.getMetric(MetricNames.ACC_TASK_BUSY_TIME) != null) {
./flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/util/MutableIOMetrics.java:                                        metrics.getMetric(MetricNames.ACC_TASK_BUSY_TIME));
./flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricNames.java:    public static final String TASK_BUSY_TIME = "busyTimeMs" + SUFFIX_RATE;
./flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricNames.java:    public static final String ACC_TASK_BUSY_TIME = "accumulateBusyTimeMs";
./flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskIOMetricGroup.java:        this.busyTimePerSecond = gauge(MetricNames.TASK_BUSY_TIME, this::getBusyTimePerSecond);
./flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskIOMetricGroup.java:                gauge(MetricNames.ACC_TASK_BUSY_TIME, this::getAccumulatedBusyTime);
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(task0, MetricNames.TASK_BUSY_TIME, "0"));
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(task1, MetricNames.TASK_BUSY_TIME, "900"));
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(task3, MetricNames.TASK_BUSY_TIME, "700"));
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(task0, MetricNames.TASK_BUSY_TIME, "0"));
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(speculativeTask0, MetricNames.TASK_BUSY_TIME, "800"));
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(task1, MetricNames.TASK_BUSY_TIME, "900"));
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(speculativeTask1, MetricNames.TASK_BUSY_TIME, "100"));
./flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java:        dumps.add(new GaugeDump(task3, MetricNames.TASK_BUSY_TIME, "700"));
Copy code
$ find . -name "*.java" | xargs grep setEnableBusyTime 2>/dev/null
./flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java:        getEnvironment().getMetricGroup().getIOMetricGroup().setEnableBusyTime(false);
./flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:            environment.getMetricGroup().getIOMetricGroup().setEnableBusyTime(true);
./flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskIOMetricGroup.java:    public void setEnableBusyTime(boolean enabled) {
./flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/TaskIOMetricGroupTest.java:        taskIO.setEnableBusyTime(true);
Copy code
private double getBusyTimePerSecond() {
    double busyTime = idleTimePerSecond.getValue() + getBackPressuredTimeMsPerSecond();
    return busyTimeEnabled ? 1000.0 - Math.min(busyTime, 1000.0) : Double.NaN;
}
setEnableBusyTime seems eing called to true for "StreamTask" and false for "SourceStreamTask"? Other than that, I quite dont see any other place where this variable is set. But it looks like it will be... false for every source.... Weird. This modern source api was implemented: https://github.com/apache/beam/issues/25486
Maybe you may find this useful, as it should affect your case too i believe
but its kind of weird still, busy should just be reversed idle. and need to track this code too...
if we are both interested in source scaling too, maybe we should also add compilation of apache beam on top πŸ˜„
and importing another patch like this
if we get then idle stat for source, then ifbusy is still N/A, maybe it may be worth to check this setting... hopefully will not be needed
v
So.. as you predicted, I am now facing the same issue πŸ˜… I got sidetracked while trying to use the Flink reactive mode for scaling, but that didnt work with Batch jobs (I am investigating scaling out batch jobs). With the operator autoscaler, the problem I faced is similar to yours: None of my vertices are showing any Busy values 😞 (all nodes are showing N/A busy values). I haven't been able to debug it thus far, though thanks for your code pointers, they were highly relevant πŸ™‡ I indeed suspect it might be either due to: me using batch job instead of streaming job, or due the multiple custom sources sources. Will update here if I find anything..
Changed my beam job to streaming mode, and now I am indeed getting busy metrics for all the nodes (EXCEPT the source node as you predicted) in the exact same configuration. Unfortunately the problem is even worse in batch job where none of the nodes show the busy values, I still havent been able to debug why πŸ€”
So I noticed that the Beam Flink runner is STILL using the old deprecated DataSet API for batch jobs, and the busy time metrics are supported on the unified DataStream API which supports both batch and stream modes. I see https://github.com/apache/beam/pull/25742 never got reviewed or merged, even though the first change which lays the groundwork for the DataSet -> DataStream migration got merged: https://github.com/apache/beam/pull/25525 cc: @Becket Qin β€’ Were you able to successfully use the DataStream API for batch jobs in production? β€’ As discussed above, were you able to get % busy metrics for source vertices as well? We see https://github.com/apache/beam/issues/25741 is also a change which is open and might be related
b
Yes, we have run the DataStream based batch runner against some production data. https://github.com/apache/beam/pull/25742 was merged to the LinkedIn fork.
And the % busy metrics is available for source vertices.
v
Wow this is really helpful, thanks for the pointers (and for your contributions to Beam+Flink πŸ™Œ )
To clarify regarding the % busy metrics for sources (which doesnt work today in either streaming or batch): is https://github.com/apache/beam/pull/25742 enough or do we also need https://github.com/apache/beam/issues/25741?
a
Thanks for info back! I had too many problems with flink actually with autoscaler. TBH, source scaling is actually less of an issue, because actually it can be put on single worker or 2 statistically, it does not exercse CPU much for me. But I cant make with those decisions of autoscaler, they are so weird.It has 26 individual tasks after chaining, mostly hashing elements. And then when CPU is too high it scales.... just one vertex out of 26. OF course it does little, so it makes 3-4 bad decisions to scale up and I have 2 more workers than necessary generating costs. And then it finally catches up with the work, start dropping to normal value - perfect for the job. And then suddenly it overestimates itself and drops again to below value and whole massare begins with 5-6 scaling before it reaches desired state. So typically my system lags and does not function properly. I would say, autoscaler fails completely me. I will need to find static scaling -> increase job parallelism instead of vertexes if possible. I suggest it may be way to go, especially since only from 1.18 we have appareantly job vertexes rescaling in place
plusone 1
πŸ’― 1
v
Thanks for pointing these out! I faced similar issues, glad to see I am not the only one. Another issue which I faced in particular was the main sink vertex autoscaling up to the maximum paralellism always, because it is always 100% busy when the actual bottleneck is the external system being written to (e.g. SQL DB being too busy). Due to false-positive/step-wise scaling, I was seeing frequent job interrupts because all the job pods get killed and restarted on any scaling decision 😞
a
hmm... for sink issue, sounds a bit like you may need to fix SQL DB, but also, there is an option of excluding specific vertexes from scaling all together. Altgough this "ID" seems a bit weird, unsure if you are supposed to know just UID or what exactly.... and how can you tell before jub is submitted? Are UIDs at least stable before restarts? If so you could potentially check autoscaler log, but I never used this option, so cant be sure: kubernetes.operator.job.autoscaler.vertex.exclude.ids Tbh, Im thinking of actually making some experimental rewrite to native flink API during next refactor which is scheduled for me though anyway... then 1.18 maybe really can reduce those restarts. But it is only option probably if you have written things in Java already and difficult code is engine agnostic and you just replace operators and some syntax. To some extent it actually seems true so far. Then if this does not work, I wonder if HPA can do some general scaling: https://github.com/apache/flink-kubernetes-operator/blob/main/examples/hpa/basic-hpa.yaml Horizontal Pod Autoscaler It is possible to provide the Horizontal Pod Autoscaler configuration through the yaml files. To run this example Kubernetes 1.23 or newer is needed, so the 
autoscaling/v2
is available. The feature is experimental so use it with caution but here is an example configuration:
Copy code
kubectl apply -f hpa/basic-hpa.yaml
Maybe it could potentially be worth
2 Views
Open in Slack
PreviousNext
Powered by