{"record":{"id":"99836ba5e4168198","repo":"conductor-oss/conductor","slug":"invalid-row-with-entitykey-s-found-in-datastore","errorCode":null,"errorMessage":"Invalid row with entityKey: %s found in datastore for workflow: %s","messagePattern":"Invalid row with entityKey: (.+?) found in datastore for workflow: (.+?)","errorType":"exception","errorClass":"NonTransientException","httpStatus":500,"severity":"error","filePath":"cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraExecutionDAO.java","lineNumber":535,"sourceCode":"                        session.execute(\n                                selectWorkflowWithTasksStatement.bind(\n                                        workflowUUID, DEFAULT_SHARD_ID));\n                List<TaskModel> tasks = new ArrayList<>();\n\n                List<Row> rows = resultSet.all();\n                if (rows.size() == 0) {\n                    LOGGER.info(\"Workflow {} not found in datastore\", workflowId);\n                    return null;\n                }\n                for (Row row : rows) {\n                    String entityKey = row.getString(ENTITY_KEY);\n                    if (ENTITY_TYPE_WORKFLOW.equals(entityKey)) {\n                        workflow = readValue(row.getString(PAYLOAD_KEY), WorkflowModel.class);\n                    } else if (ENTITY_TYPE_TASK.equals(entityKey)) {\n                        TaskModel task = readValue(row.getString(PAYLOAD_KEY), TaskModel.class);\n                        tasks.add(task);\n                    } else {\n                        throw new NonTransientException(\n                                String.format(\n                                        \"Invalid row with entityKey: %s found in datastore for workflow: %s\",\n                                        entityKey, workflowId));\n                    }\n                }\n\n                if (workflow != null) {\n                    recordCassandraDaoRequests(\"getWorkflow\", \"n/a\", workflow.getWorkflowName());\n                    tasks.sort(Comparator.comparingInt(TaskModel::getSeq));\n                    workflow.setTasks(tasks);\n                }\n            } else {\n                resultSet = session.execute(selectWorkflowStatement.bind(workflowUUID));\n                workflow =\n                        Optional.ofNullable(resultSet.one())\n                                .map(\n                                        row -> {\n                                            WorkflowModel wf =","sourceCodeStart":517,"sourceCodeEnd":553,"githubUrl":"https://github.com/conductor-oss/conductor/blob/cf7c3e4a8adfb158be778ab1ec525323c363cd3a/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraExecutionDAO.java#L517-L553","documentation":"While reconstructing a Workflow from its Cassandra rows, the DAO encountered a row whose ENTITY_KEY is neither ENTITY_TYPE_WORKFLOW nor ENTITY_TYPE_TASK. It throws NonTransientException because an unknown entity type signals data corruption or an unexpected schema state that re-reading will not fix. NonTransientException is NOT retried by the framework RetryTemplate, so it surfaces immediately.","triggerScenarios":"getWorkflow(workflowId) returns a partition whose rows contain an entityKey value other than the recognized workflow/task constants — e.g. a new entity type written by a newer Conductor version, a manual row insert, or a partial/failed migration.","commonSituations":"Rolling Conductor versions where a newer build writes a new ENTITY_TYPE_* that the running build does not recognize; direct CQL writes to the workflow table; schema drift between environments.","solutions":["Inspect the offending row(s) in Cassandra: SELECT entity_key, payload_key FROM <table> WHERE workflow_id = ? and identify the unexpected entityKey.","Align all Conductor nodes to the same version so ENTITY_TYPE constants match what is written.","Repair or delete the corrupt/unknown row so only WORKFLOW and TASK rows remain for that workflowId.","If a legitimate new entity type was introduced, upgrade this DAO's read path to handle it instead of throwing."],"exampleFix":"// before\n} else {\n    throw new NonTransientException(\n        String.format(\"Invalid row with entityKey: %s ...\", entityKey, workflowId));\n}\n\n// after (only if the new entity type is intentional and can be skipped/handled)\n} else if (ENTITY_TYPE_NEW_ENTITY.equals(entityKey)) {\n    LOGGER.warn(\"Skipping {} row for workflow {}\", entityKey, workflowId);\n} else {\n    throw new NonTransientException(/* ... */);\n}","handlingStrategy":"validation","validationCode":"// Pre-validate known entity types before delegating to the read path (if you own the read)\nSet<String> known = Set.of(ENTITY_TYPE_WORKFLOW, ENTITY_TYPE_TASK);\n// There is no public pre-check; this is a data-integrity guard you would add in a repair job:\n// SELECT entity_key FROM <table> WHERE workflow_id = ? and assert entity_key IN known.","typeGuard":null,"tryCatchPattern":"try {\n    Workflow w = executionDAOFacade.getWorkflow(workflowId, true);\n} catch (NonTransientException e) {\n    // data corruption — surface a 409/500 and trigger a row-repair investigation\n    LOGGER.error(\"Corrupt row for workflow {}\", workflowId, e);\n    throw e;\n}","preventionTips":["Keep all Conductor nodes on the same version so ENTITY_TYPE constants match written rows.","Never write to the workflow table with CQL manually; use the DAO.","Run a row-entityKey audit after major version upgrades."],"tags":["cassandra","persistence","data-integrity","non-transient","schema","dao"],"backgroundTag":null,"analyzedSha":"cf7c3e4a8adfb158be778ab1ec525323c363cd3a","analyzedAt":"2026-08-14T03:33:19.897Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}