{"record":{"id":"2675ec8c7d1c14d3","repo":"apache/pulsar","slug":"e-getmessage-2675ec","errorCode":null,"errorMessage":"e.getMessage()","messagePattern":"e\\.getMessage\\(\\)","errorType":"http","errorClass":"RestException","httpStatus":500,"severity":"error","filePath":"pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java","lineNumber":1225,"sourceCode":"                    MessageId newMsgId = MessageId.fromByteArray(\n                            Base64.getDecoder().decode((String) msg.getProperties().get(\"__pfn_input_msg_id__\")));\n\n                    if (msgId.equals(newMsgId)\n                            && msg.getProperties().get(\"__pfn_input_topic__\")\n                            .equals(TopicName.get(inputTopicToWrite).toString())) {\n                        return new String(msg.getData());\n                    }\n                }\n                curTime = System.currentTimeMillis();\n            }\n            throw new RestException(Status.REQUEST_TIMEOUT, \"Request Timed Out\");\n        } catch (SchemaSerializationException e) {\n            throw new RestException(Status.BAD_REQUEST, String.format(\n                    \"Failed to serialize input with error: %s. Please check\"\n                            + \"if input data conforms with the schema of the input topic.\",\n                    e.getMessage()));\n        } catch (IOException e) {\n            throw new RestException(Status.INTERNAL_SERVER_ERROR, e.getMessage());\n        } finally {\n            if (reader != null) {\n                reader.closeAsync();\n            }\n            if (producer != null) {\n                producer.closeAsync();\n            }\n        }\n    }\n\n    @Override\n    public FunctionState getFunctionState(final String tenant,\n                                          final String namespace,\n                                          final String functionName,\n                                          final String key,\n                                          final AuthenticationParameters authParams) {\n\n        if (!isWorkerServiceAvailable()) {","sourceCodeStart":1207,"sourceCodeEnd":1243,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java#L1207-L1243","documentation":"During a function trigger (test message) the worker serializes your input into the input topic's schema using an AUTO_PRODUCE_BYTES producer. If the payload cannot be encoded with that schema, a SchemaSerializationException is wrapped into a 400 RestException with the schema error appended. An IOException on the underlying producer/reader operations is instead converted to a 500 RestException carrying the raw exception message.","triggerScenarios":"Calling PUT /admin/v3/functions/{tenant}/{namespace}/{functionName}/trigger with input bytes that do not conform to the declared schema of the function's input topic (at line 1225 specifically: any IOException from producer.send/reader operations in triggerFunction).","commonSituations":"Sending JSON text to a topic with an Avro/Protobuf schema; sending a string to an int64-schema topic; schema evolved on the topic but the client payload was not updated; broker/bookkeeper connectivity problems producing IOExceptions during send.","solutions":["Match the trigger payload to the input topic's declared schema (e.g. send valid Avro/JSON-encoded bytes for an AvroSchema topic).","Inspect the appended e.getMessage() to see which field or encoding the schema rejected.","For 500/IOException variants, check worker-to-broker connectivity and topic existence, then retry the trigger.","Verify the function's input schema config (or use AUTO_CONSUME / correct schema type) if the schema was set incorrectly at registration."],"exampleFix":"// before: trigger with raw JSON string against an Avro-schema topic\ncurl -X PUT .../trigger --data '{\"id\":\"not-a-long\"}'\n\n// after: send schema-conformant payload (or base64 Avro-encoded bytes)\ncurl -X PUT .../trigger --data '{\"id\":42}'","handlingStrategy":"validation","validationCode":"// Validate payload against the topic schema before triggering (pseudocode using Pulsar client)\nSchema<T> schema = ...; // same schema as the input topic\ntry {\n    byte[] encoded = schema.encode(payload); // throws SchemaSerializationException if incompatible\n} catch (SchemaSerializationException e) {\n    // fix payload before calling the trigger endpoint\n}","typeGuard":null,"tryCatchPattern":"try {\n    Response r = triggerFunction(tenant, ns, fn, payload);\n} catch (RestApiException e) {\n    if (e.getResponse().getStatus() == 400 && e.getMessage().contains(\"Failed to serialize input\")) {\n        // adjust payload to schema\n    } else if (e.getResponse().getStatus() == 500) {\n        // IOException: check broker connectivity, retry\n    }\n}","preventionTips":["Keep trigger payloads in sync with the input topic's schema, especially after schema evolution.","Test schema encoding locally with the same Schema instance the topic uses.","Check broker connectivity/health before running trigger-based smoke tests."],"tags":["rest-api","schema","functions","serialization"],"backgroundTag":"schema-serialization-failed","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}