{"record":{"id":"114344e4415a420b","repo":"zhisheng17/flink-learning","slug":"this-database-database-does-not-exist","errorCode":null,"errorMessage":"This ${database} database does not exist!","messagePattern":"This (.+?) database does not exist!","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink-learning-connectors/flink-learning-connectors-influxdb/src/main/java/com/zhisheng/connectors/influxdb/InfluxDBSink.java","lineNumber":42,"sourceCode":"\n    private final InfluxDBConfig influxDBConfig;\n\n    public InfluxDBSink(InfluxDBConfig influxDBConfig) {\n        this.influxDBConfig = Preconditions.checkNotNull(influxDBConfig, \"InfluxDB client config should not be null\");\n    }\n\n    @Override\n    public void open(Configuration parameters) throws Exception {\n        super.open(parameters);\n\n        influxDBClient = InfluxDBFactory.connect(influxDBConfig.getUrl(), influxDBConfig.getUsername(), influxDBConfig.getPassword());\n\n        if (!influxDBClient.databaseExists(influxDBConfig.getDatabase())) {\n            if(influxDBConfig.isCreateDatabase()) {\n                influxDBClient.createDatabase(influxDBConfig.getDatabase());\n            }\n            else {\n                throw new RuntimeException(\"This \" + influxDBConfig.getDatabase() + \" database does not exist!\");\n            }\n        }\n\n        influxDBClient.setDatabase(influxDBConfig.getDatabase());\n\n        if (influxDBConfig.getBatchActions() > 0) {\n            influxDBClient.enableBatch(influxDBConfig.getBatchActions(), influxDBConfig.getFlushDuration(), influxDBConfig.getFlushDurationTimeUnit());\n        }\n\n        if (influxDBConfig.isEnableGzip()) {\n            influxDBClient.enableGzip();\n        }\n    }\n\n    @Override\n    public void invoke(MetricEvent metricEvent, Context context) throws Exception {\n        if (StringUtils.isNullOrWhitespaceOnly(metricEvent.getName())) {\n            throw new RuntimeException(\"No measurement defined\");","sourceCodeStart":24,"sourceCodeEnd":60,"githubUrl":"https://github.com/zhisheng17/flink-learning/blob/d731cee7618021be56d132cc925102ffff8d75e6/flink-learning-connectors/flink-learning-connectors-influxdb/src/main/java/com/zhisheng/connectors/influxdb/InfluxDBSink.java#L24-L60","documentation":"InfluxDBSink.open checks whether the configured database exists in InfluxDB before writing. If it does not exist and influxDBConfig.isCreateDatabase() is false, the sink fails fast with a RuntimeException naming the missing database.","triggerScenarios":"Opening the InfluxDBSink connected to an InfluxDB instance where the database named in InfluxDBConfig.getDatabase() does not exist, with the createDatabase flag left at false.","commonSituations":"Deploying to a fresh InfluxDB instance without creating the database, typos in the database name in config, or environments where CREATE DATABASE requires privileges the sink does not have so auto-creation was disabled.","solutions":["Create the database in InfluxDB first: CREATE DATABASE \"mydb\" (or HTTP /query?q=CREATE DATABASE%20\"mydb\")","Enable auto-creation by setting createDatabase(true) on InfluxDBConfig","Fix the database name typo in the sink config"],"exampleFix":"// before\nInfluxDBConfig config = new InfluxDBConfig.Builder(\"http://localhost:8086\", \"metrics\").build();\n// after\nInfluxDBConfig config = new InfluxDBConfig.Builder(\"http://localhost:8086\", \"metrics\")\n    .createDatabase(true)\n    .build();","handlingStrategy":"fallback","validationCode":"// before creating the sink\ntry (InfluxDB influxDB = InfluxDBFactory.connect(url)) {\n    if (!influxDB.databaseExists(database)) {\n        influxDB.createDatabase(database);\n    }\n}","typeGuard":null,"tryCatchPattern":"try {\n    sink.open(ctx);\n} catch (RuntimeException e) {\n    if (e.getMessage() != null && e.getMessage().contains(\"database does not exist\")) {\n        influxDB.createDatabase(config.getDatabase());\n        sink.open(ctx);\n    } else throw e;\n}","preventionTips":["Enable createDatabase(true) in InfluxDBConfig for non-production/fresh environments","Provision the database via IaC or init scripts before deploying the Flink job","Verify the database name in config matches the one created in InfluxDB"],"tags":["influxdb","database","flink","config"],"backgroundTag":"resource-not-found","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"}