{"record":{"id":"f7a4feab7a433a54","repo":"apache/seatunnel","slug":"invalid-row","errorCode":"INVALID_ROW","errorMessage":"Input row arity ${arity} does not match the configured schema arity ${rowArity}","messagePattern":"Input row arity (.+?) does not match the configured schema arity (.+?)","errorType":"error_code","errorClass":"SnmpConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-snmp/src/main/java/org/apache/seatunnel/connectors/seatunnel/snmp/sink/SnmpSinkRowConverter.java","lineNumber":80,"sourceCode":"    private static final BigInteger TICKS_PER_MINUTE = BigInteger.valueOf(6_000L);\n    private static final BigInteger TICKS_PER_SECOND = BigInteger.valueOf(100L);\n\n    private final int rowArity;\n    private final int oidIndex;\n    private final int valueIndex;\n    private final int valueTypeIndex;\n\n    SnmpSinkRowConverter(SnmpSinkConfig config, SeaTunnelRowType rowType) {\n        this.rowArity = rowType.getTotalFields();\n        this.oidIndex = requireStringField(rowType, config.getOidField(), \"oid_field\");\n        this.valueIndex = requireStringField(rowType, config.getValueField(), \"value_field\");\n        this.valueTypeIndex =\n                requireStringField(rowType, config.getValueTypeField(), \"value_type_field\");\n    }\n\n    SnmpSetRequest convert(SeaTunnelRow row) {\n        if (row.getArity() != rowArity) {\n            throw invalidRow(\n                    \"Input row arity \"\n                            + row.getArity()\n                            + \" does not match the configured schema arity \"\n                            + rowArity);\n        }\n\n        String oid = requireNonBlankRowValue(row, oidIndex, \"OID\");\n        String value = requireNonNullRowValue(row, valueIndex, \"value\");\n        String valueType = requireNonBlankRowValue(row, valueTypeIndex, \"value type\");\n        return new SnmpSetRequest(parseOid(oid), parseVariable(valueType, value));\n    }\n\n    private static int requireStringField(\n            SeaTunnelRowType rowType, String fieldName, String optionName) {\n        int index = rowType.indexOf(fieldName, false);\n        if (index < 0) {\n            throw invalidConfig(\n                    \"Option `\"","sourceCodeStart":62,"sourceCodeEnd":98,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-snmp/src/main/java/org/apache/seatunnel/connectors/seatunnel/snmp/sink/SnmpSinkRowConverter.java#L62-L98","documentation":"The SNMP sink validates each incoming SeaTunnelRow against the schema arity declared in the sink configuration before performing any network I/O. If the row's field count differs from the configured row type, conversion is rejected with INVALID_ROW to prevent index misalignment when reading the oid/value fields.","triggerScenarios":"SnmpSinkRowConverter.convert(row) is called (from the sink's write path) with a SeaTunnelRow whose getArity() != rowArity — e.g. an upstream transform changed the number of columns, or the sink table schema was edited without updating the upstream source.","commonSituations":"Adding/removing a column in the source SQL or transform upstream of the SNMP sink; two sinks sharing one dataset with different schemas; row produced by a custom Source with a stale SeaTunnelRowType.","solutions":["Align the upstream schema so the row arity matches the sink's configured SeaTunnelRowType","Check the source/transform column list and remove or add columns to match the SNMP sink table definition","Rebuild the pipeline after schema edits — old checkpoints may replay rows with the old arity, so restart with a fresh state if needed","Wrap the sink write path in a catalog/schema consistency check at job submit time"],"exampleFix":"// before: source SELECT with 3 cols, sink schema with 2\nSELECT host, oid, value FROM events\n// after\nSELECT oid, value FROM events  -- matches sink schema arity 2","handlingStrategy":"validation","validationCode":"// Before the sink write, assert arity:\nif (row.getArity() != sinkRowType.getTotalFields()) {\n    throw new IllegalArgumentException(\"row arity \" + row.getArity()\n        + \" != schema arity \" + sinkRowType.getTotalFields());\n}","typeGuard":null,"tryCatchPattern":"try {\n    sink.write(row);\n} catch (SeaTunnelException e) {\n    if (e.getMessage().contains(\"does not match the configured schema arity\")) {\n        LOG.error(\"Upstream schema drift detected for SNMP sink\", e);\n    }\n    throw e;\n}","preventionTips":["Keep sink table schemas and upstream source/transform column lists in one reviewed place","Re-run schema validation after every pipeline edit","Avoid sharing one dataset between sinks with different schemas","Restart from clean state after schema changes to avoid replayed old-arity rows"],"tags":["snmp","sink","schema","row-validation"],"backgroundTag":"schema-validation-failed","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}