{"record":{"id":"8fa97c5c930c405c","repo":"zhisheng17/flink-learning","slug":"pwd-env-doesn-t-contains-yarn-application-id-or-co","errorCode":null,"errorMessage":"PWD env doesn't contains yarn application id or container id","messagePattern":"PWD env doesn't contains yarn application id or container id","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink-learning-extends/flink-metrics/flink-metrics-kafka/src/main/java/org/apache/flink/metrics/kafka/KafkaReporter.java","lineNumber":79,"sourceCode":"\n\t@Override\n\tpublic void open(MetricConfig config) {\n\t\tMap<String, String> envs = System.getenv();\n\t\tString clusterId = envs.get(\"CLUSTER_ID\");\n\t\tif (clusterId != null) {\n\t\t\t//k8s cluster\n\t\t\tappId = clusterId;\n\t\t\tcontainerId = envs.get(\"HOSTNAME\");\n\t\t} else {\n\t\t\t//yarn cluster\n\t\t\tString pwd = envs.get(\"PWD\");\n\t\t\tString[] values = pwd.split(File.separator);\n\t\t\tif (values.length >= 2) {\n\t\t\t\tappId = values[values.length - 2];\n\t\t\t\tcontainerId = values[values.length - 1];\n\t\t\t} else {\n\t\t\t\tLOG.error(\"PWD env ({}) doesn't contains yarn application id or container id\", pwd);\n\t\t\t\tthrow new RuntimeException(\n\t\t\t\t\t\"PWD env doesn't contains yarn application id or container id\");\n\t\t\t}\n\t\t}\n\n\t\tProperties properties = System.getProperties();\n\t\ttaskName = properties.getProperty(\"taskName\", null);\n\t\ttaskId = properties.getProperty(\"taskId\", null);\n\n\t\tProperties props = new Properties();\n\t\tString clientIdPrefix = taskId != null ? taskId : appId;\n\t\tprops.put(\"client.id\", \"flink_\" + clientIdPrefix + \"_metrics\");\n\t\tprops.put(\"bootstrap.servers\", getString(config, BOOTSTRAP_SERVERS));\n\t\tprops.put(\"acks\", getString(config, ACKS));\n\t\tprops.put(\"retries\", getInteger(config, RETRIES));\n\t\tprops.put(\"batch.size\", getInteger(config, BATCH_SIZE));\n\t\tprops.put(\"linger.ms\", getInteger(config, LINGER_MS));\n\t\tprops.put(\"buffer.memory\", getInteger(config, BUFFER_MEMORY));\n\t\tprops.put(\"max.request.size\", getInteger(config, MAX_REQUEST_SIZE));","sourceCodeStart":61,"sourceCodeEnd":97,"githubUrl":"https://github.com/zhisheng17/flink-learning/blob/d731cee7618021be56d132cc925102ffff8d75e6/flink-learning-extends/flink-metrics/flink-metrics-kafka/src/main/java/org/apache/flink/metrics/kafka/KafkaReporter.java#L61-L97","documentation":"KafkaReporter (Flink metrics) derives the YARN application id and container id by splitting the current working directory (PWD of the YARN container). If PWD has fewer than two path segments it cannot extract both ids, logs an error, and throws RuntimeException in open().","triggerScenarios":"Running the reporter outside a YARN container (local IDE run, standalone/minikube deployment) where PWD is '/' or a shallow path, so pwd.split(File.separator).length < 2.","commonSituations":"Testing the job locally with the Kafka metrics reporter enabled; running on Kubernetes or a non-YARN cluster; container working directory layout changed by platform upgrade.","solutions":["Run the job on YARN so the container PWD contains the application/container id path segments.","Disable or swap the KafkaReporter for local/non-YARN runs (use a different reporter in local profile).","Patch the reporter to take appId/containerId from env vars (FLINK_APPLICATION_ID, HOSTNAME) as a fallback instead of PWD.","Guard the reporter initialization with a check for YARN before throwing."],"exampleFix":"// before\nString[] values = pwd.split(File.separator);\nif (values.length >= 2) { ... } else { throw new RuntimeException(...); }\n// after\nString[] values = pwd.split(File.separator);\nif (values.length >= 2) { ... }\nelse if (System.getenv(\"FLINK_APPLICATION_ID\") != null) {\n    appId = System.getenv(\"FLINK_APPLICATION_ID\");\n    containerId = System.getenv(\"HOSTNAME\");\n} else { throw new RuntimeException(...); }","handlingStrategy":"try-catch","validationCode":"String pwd = System.getProperty(\"user.dir\");\nString[] segs = pwd.split(java.io.File.separatorChar == '\\\\' ? \"\\\\\\\\\" : \"/\");\nboolean yarnLike = segs.length >= 2 && segs[segs.length-2].startsWith(\"application_\");\nif (!yarnLike) disableKafkaMetricsReporter();","typeGuard":null,"tryCatchPattern":"try {\n    reporter.open(metrics);\n} catch (RuntimeException e) {\n    LOG.warn(\"KafkaReporter cannot start (no YARN PWD): {}\", e.getMessage());\n    // fall back to another reporter\n}","preventionTips":["Only enable KafkaReporter on YARN deployments","Use separate reporter config for local vs cluster runs","Source appId/containerId from env vars with PWD as fallback","Log clearly when the reporter is skipped due to environment"],"tags":["flink","metrics","yarn","environment"],"backgroundTag":"missing-env-var","analyzedSha":"d731cee7618021be56d132cc925102ffff8d75e6","analyzedAt":"2026-09-06T05:35:08.496Z","contentChangedAt":"2026-09-06T05:35:08.496Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}