Troubleshooting Apache Flink jobs

org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: 
Could not allocate all requires slots within timeout of 300000 ms. 
Slots required: 8, slots allocated: 0

The Flink web interface is accessible and in the overview page, you see 0 (zero) as the number of available task slots.

• If this issue happens after initial configuration, it means that you did not configure enough task slots for running all the jobs. Verify whether the number of task slots that are displayed in the Flink web interface is equal to, or greater than, the number of running jobs. If it is not, update your IBM Business Automation Insights configuration with the correct number of task manager replicas and task slots.
• If the issue happens after you have updated your Business Automation Insights configuration, the problem might indicate that Flink did not correctly update the metadata about task slot assignment after a failing task manager recovered.

1. Restart each task manager one by one, in any order, by running these delete commands. The <custom_resource_name> placeholder stands for the name of your custom resource for IBM Business Automation Insights.

   oc delete pod <custom_resource_name>-[...]-ep-taskmanager-0
   oc delete pod <custom_resource_name>-[...]-ep-taskmanager-1
   ...

   These commands redistribute the jobs to the different task slots. The task manager StatefulSet immediately redeploys new instances.

2. After the restart, verify from the Flink web interface that all jobs are running and that task slots are correctly assigned.

Get latest completed checkpoint for <job-id> job
REST endpoint did not return latest completed checkpoint. Getting it from Persistent Volume...
Error: There is no checkpoint to recover from.

1. You can find the latest successful checkpoint in the <bai-pv>/checkpoints/<job-name>/<job-id> directory.

   <bai-pv>

   The directory where the Business Automation Insights persistent volume (PV) was created. Set this variable to /mnt/pv, which is the folder where the PV is mapped within the job submitters.

   <job-name>

   The name of the failing job, for example bai/bpmn.

   <job-id>

   The job identifier that is indicated by the error message. Pick the most recent checkpoint, that is, the higher <checkpoint-id> value and verify that the folder is not empty.

2. If all <checkpoint-id> folders are empty, and only in this case, use the latest savepoint of the corresponding processing job, which you can find in the <bai-pv>/savepoints/<job-name> directory.
3. Update the <job_name>.recovery_path parameter of the failing job by following the procedure in Updating your Business Automation Insights custom resource.

{"errors":["Service temporarily unavailable due to an ongoing leader election. 
Please refresh."]}

INFO org.apache.flink.runtime.jobmaster.slotpool.SlotPool
- Cannot serve slot request, no ResourceManager connected. Adding as pending request [SlotRequestId{######}]
...
INFO  org.apache.flink.runtime.jobmaster.slotpool.SlotPool   
- Pending slot request [SlotRequestId{######}] timed out.

org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: 
Could not allocate all requires slots within timeout of 300000 ms. 
Slots required: 6, slots allocated: 0

oc delete pod <custom_resource_name>-[...]-ep-jobmanager-<id>

A new job manager instance is deployed. After the redeployment, all jobs should be running again and the Flink web interface should be accessible.

• The Flink pods for job manager or task managers are unhealthy (error status or stuck in Init state).
• The Flink web interface is accessible but one or several Flink jobs that should be deployed are not shown as running. Either they are absent or the Flink web interface reports errors for their task managers, and these errors are different from the other entries in this page.
• No events reach Business Performance Center.

1. Retrieve the names of the jobs. The full list of jobs is bai-bpmn, bai_bawadv, bai-icm, bai-odm, bai-content, iaf-insights-engine-event-forwarder.

   oc get jobs -l 'app.kubernetes.io/name in (ibm-business-automation-insights,iaf-insights-engine)' -o=custom-columns=Name:.metadata.name --no-headers

   The returned list might be only a subset of the full list because the deployment of these jobs is conditional and depends on the activation of event processors in the custom resource.

2. From the list that this command returns, select the subset that is actually present in the configuration of Business Automation Insights event processing. In this example, all event processors are activated in the custom resource, hence the subset is identical to the full list.

   oc get job icp4adeploy-bai-content -o json | jq 'del(.spec.selector)' | jq 'del(.spec.template.metadata.labels)' | oc replace --force -f -
   oc get job icp4adeploy-bai-bpmn -o json | jq 'del(.spec.selector)' | jq 'del(.spec.template.metadata.labels)' | oc replace --force -f -
   oc get job icp4adeploy-bai-bawadv -o json | jq 'del(.spec.selector)' | jq 'del(.spec.template.metadata.labels)' | oc replace --force -f -
   oc get job icp4adeploy-bai-icm -o json | jq 'del(.spec.selector)' | jq 'del(.spec.template.metadata.labels)' | oc replace --force -f -
   oc get job icp4adeploy-bai-odm -o json | jq 'del(.spec.selector)' | jq 'del(.spec.template.metadata.labels)' | oc replace --force -f -
   oc get job iaf-insights-engine-event-forwarder -o json | jq 'del(.spec.selector)' | jq 'del(.spec.template.metadata.labels)' | oc replace --force -f -
   

3. For each of the job names , run the following command by replacing the <job_name> placeholder with the actual name of the job.

   oc get job <job_name> -o json | jq 'del(.spec.selector)' | jq 'del(.spec.template.metadata.labels)' | oc replace --force -f -

1. When you run the get pods command, you might observe the following results.

   oc get pods -n <bai-namespace>

   Table 1. Pod retrieval results

   NAME	READY	RESTARTS	STATUS	AGE
   ... <bai_cr_name>-bpmn-aaaaa ... <bai_cr_name>-bai-icm-bbbbb ...	... 0/1 ... 0/1 ...	... Init:0/1 ... Init:0/1 ...	... 0 ... 0 ...	... 2h ... 2h ...

2. Retrieve the job manager logs and look up the logs of the <custom_resource_name>-bai-bpmn and <custom_resource_name>-bai-icm pods.

   JOB_MANAGER=$(oc get pod -l component=jobmanager --no-headers -o custom-columns=NAME:.metadata.name)
   oc logs $JOB_MANAGER -c jobmanager

1. Delete all the pods that were previously created by the <custom_resource_name>-bai-setup job.

   oc delete pod <custom_resource_name>-bai-setup-aaaaa
   oc delete pod <custom_resource_name>-bai-setup-bbbbb
   oc delete pod <custom_resource_name>-bai-setup-ccccc
   oc delete pod <custom_resource_name>-bai-setup-ddddd

2. Run the following command to re-create the <bai_cr_name>-bai-setup job.

   oc get job <custom_resource_name>-bai-setup -o json | jq "del(.spec.selector)" | jq "del(.spec.template.metadata.labels)" | oc replace --force -f -

Caused by: java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint file:/mnt/pv/savepoints/dba/bai-<job-name>/savepoint-<savepoint-id> Cannot map checkpoint/savepoint state for operator xxxxxxxxxxxxxx to the new program, because the operator is not available in the new program.

1. Enable the option for verbose logs in the custom resource (CR) YAML file.

   spec.bai_configuration.flink.verbose_logs: true

2. Restart the job manager.

   oc delete pod <jobmanager> -n <bai_namespace>

3. Restart all the task managers.

   oc delete pod <taskmanager> -n <bai_namespace>

4. Restart the Flink jobs.
   1. Retrieve the names of the jobs.

      oc get jobs -o=custom-columns=Name:.metadata.name --no-headers

   2. For each job name, run the following command by replacing the <job_name> placeholder with the actual name of the job.

      oc get job <job_name> -o json | jq 'del(.spec.selector)' | jq 'del(.spec.template.metadata.labels)' | oc replace --force -f -

   Whenever a Flink job receives an IBM Cloud Pak for Business Automation raw event to process to a time series and possibly afterward to a summary, this raw event is written as INFO log level to the log of the task manager this Flink job is deployed to.

1. Identify the latest checkpoints that were written by the job manager for each Flink job before the outage.
   1. Connect to the job manager.

      JOB_MANAGER_POD=$(oc get pods -l component=jobmanager --no-headers -o custom-columns=Namae:.metadata.name | head -n 1)
        oc rsh -c jobmanager ${JOB_MANAGER_POD}

   2. Identify the latest Flink job identifiers.

      LATEST_JOB_IDS=$(for DIR in $(find /mnt/pv/checkpoints/dba/* -type d -prune); do echo -n $DIR/; ls -t $DIR | head -n 1 | tail -n 1;done)

   3. Retrieve the directory paths containing the latest checkpoints that were written by the job manager for each running FLink job.

      for CHK_DIR in ${LATEST_JOB_IDS}; do ls -t $CHK_DIR/chk*/_metadata 2>/dev/null| head -n 1 | tail -n 1; done | sed 's#/_metadata$##'

2. Restart the Flink jobs from their respective checkpoints.
   1. Edit the Business Automation Insights custom resource for each Flink job. Example for the BPMN Flink job:

      spec:
        bai_configuration:
          bpmn:
            install: true
            parallelism: 1
            recovery_path: /mnt/pv/checkpoints/dba/bai-bpmn/6d134e8ce753f2a0ec63222532145fef/chk-67

   2. Save the changes.
   3. Check whether the Flink jobs are restarted.

      oc exec -c jobmanager -it ${JOB_MANAGER_POD} -- /opt/flink/bin/flink list | grep RUNNING