{"record":{"id":"b7465c503a2b7b52","repo":"apache/shardingsphere","slug":"unknown-roweventtype-s-b7465c","errorCode":null,"errorMessage":"Unknown rowEventType: %s","messagePattern":"Unknown rowEventType: (.+?)","errorType":"exception","errorClass":"IngestException","httpStatus":null,"severity":"error","filePath":"kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/decode/TestDecodingPlugin.java","lineNumber":77,"sourceCode":"        } else {\n            result = \"table\".equals(type) ? readTableEvent(data) : new PlaceholderEvent();\n        }\n        result.setLogSequenceNumber(logSequenceNumber);\n        return result;\n    }\n    \n    private String readEventType(final ByteBuffer data) {\n        return readNextSegment(data);\n    }\n    \n    private AbstractRowEvent readTableEvent(final ByteBuffer data) {\n        String tableName = readTableName(data);\n        String rowEventType = readRowEventType(data);\n        PipelineSQLOperationType type;\n        try {\n            type = PipelineSQLOperationType.valueOf(rowEventType);\n        } catch (final IllegalArgumentException ex) {\n            throw new IngestException(\"Unknown rowEventType: \" + rowEventType);\n        }\n        AbstractRowEvent result;\n        switch (type) {\n            case INSERT:\n                result = readWriteRowEvent(data);\n                break;\n            case UPDATE:\n                result = readUpdateRowEvent(data);\n                break;\n            case DELETE:\n                result = readDeleteRowEvent(data);\n                break;\n            default:\n                throw new IngestException(\"Unknown rowEventType: \" + rowEventType);\n        }\n        String[] tableMetaData = tableName.split(\"\\\\.\");\n        result.setSchemaName(tableMetaData[0]);\n        result.setTableName(tableMetaData[1].substring(0, tableMetaData[1].length() - 1));","sourceCodeStart":59,"sourceCodeEnd":95,"githubUrl":"https://github.com/apache/shardingsphere/blob/e952770a215630a3659c75d64369168cd3e26b82/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/decode/TestDecodingPlugin.java#L59-L95","documentation":"TestDecodingPlugin (PostgreSQL incremental ingest using the test_decoding output format) reads the table event's operation token and converts it with PipelineSQLOperationType.valueOf; an unrecognized token throws IngestException('Unknown rowEventType: <token>'). The expected tokens are the PostgreSQL test_decoding op codes (INSERT/UPDATE/DELETE); anything else — message-protocol changes, BEGIN/COMMIT tags leaking into row parsing, or plugin-format drift — fails fast.","triggerScenarios":"Running PostgreSQL logical replication with test_decoding where a row message carries an operation token that valueOf cannot resolve: malformed stream desynchronization (wrong segment-reading order), a different decoding plugin actually configured (wal2json output parsed as test_decoding), or an unexpected message type.","commonSituations":"Replication slot created with wal2json/pgoutput but the pipeline configured for test_decoding (or vice versa); stream desync after a dropped connection, so readNextSegment returns garbage tokens; PostgreSQL version differences in test_decoding output; messages for non-row operations parsed as row events.","solutions":["Verify the replication slot's plugin matches what the pipeline expects: recreate the slot with test_decoding if the job uses TestDecodingPlugin.","Inspect the logged rowEventType token; if it is BEGIN/COMMIT or JSON, the stream format does not match this decoder — fix plugin/config alignment.","Recreate the replication slot and restart incremental ingestion from a consistent position after desynchronization.","Prefer a supported, version-matched decoding setup per the ShardingSphere pipeline documentation for your PostgreSQL version."],"exampleFix":"-- before: slot uses wal2json but pipeline expects test_decoding\nSELECT * FROM pg_create_logical_replication_slot('slot1', 'wal2json');\n\n-- after\nSELECT * FROM pg_create_logical_replication_slot('slot1', 'test_decoding');","handlingStrategy":"validation","validationCode":"-- preflight: slot plugin must match the decoder the pipeline uses\nSELECT slot_name, plugin FROM pg_replication_slots WHERE slot_name = 'slot1';\n-- expected: plugin = 'test_decoding' when TestDecodingPlugin is configured","typeGuard":null,"tryCatchPattern":"try {\n    WALEvent event = plugin.decode(buffer);\n} catch (final IngestException ex) {\n    if (ex.getMessage().startsWith(\"Unknown rowEventType\")) {\n        // slot/decoder mismatch or desync: recreate slot with correct plugin, restart from snapshot\n    }\n}","preventionTips":["Create logical replication slots with exactly the plugin your pipeline config expects.","Never switch decoding plugins on an existing slot; drop and recreate the slot instead.","After connection drops, validate the next decoded token before resuming full-speed ingestion."],"tags":["pipeline","postgresql","wal","incremental","logical-decoding"],"backgroundTag":null,"analyzedSha":"e952770a215630a3659c75d64369168cd3e26b82","analyzedAt":"2026-08-14T13:54:53.392Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}