{"action":"created","issue":{"url":"https://api.github.com/repos/confluentinc/schema-registry/issues/385","repository_url":"https://api.github.com/repos/confluentinc/schema-registry","labels_url":"https://api.github.com/repos/confluentinc/schema-registry/issues/385/labels{/name}","comments_url":"https://api.github.com/repos/confluentinc/schema-registry/issues/385/comments","events_url":"https://api.github.com/repos/confluentinc/schema-registry/issues/385/events","html_url":"https://github.com/confluentinc/schema-registry/issues/385","id":166901850,"number":385,"title":"Reflection-based avro serialization is not supported","user":{"login":"phaas","id":1450137,"avatar_url":"https://avatars.githubusercontent.com/u/1450137?v=3","gravatar_id":"","url":"https://api.github.com/users/phaas","html_url":"https://github.com/phaas","followers_url":"https://api.github.com/users/phaas/followers","following_url":"https://api.github.com/users/phaas/following{/other_user}","gists_url":"https://api.github.com/users/phaas/gists{/gist_id}","starred_url":"https://api.github.com/users/phaas/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/phaas/subscriptions","organizations_url":"https://api.github.com/users/phaas/orgs","repos_url":"https://api.github.com/users/phaas/repos","events_url":"https://api.github.com/users/phaas/events{/privacy}","received_events_url":"https://api.github.com/users/phaas/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":1,"created_at":"2016-07-21T19:45:12Z","updated_at":"2016-07-21T19:46:47Z","closed_at":null,"body":"The `AbstractKafkaAvroSerializer` only handles `SpecificRecord` and `GenericRecord`s. There is no way to send a properly annotated POJO.\r\n\r\nI've tried to work around this by using a `NonRecordContainer` with raw avro-byte[]'s, but that is not a supported option either (The serializer will re-encode the byte[]s but send it using the schema of the NonRecordContainer.)\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/confluentinc/schema-registry/issues/comments/234362843","html_url":"https://github.com/confluentinc/schema-registry/issues/385#issuecomment-234362843","issue_url":"https://api.github.com/repos/confluentinc/schema-registry/issues/385","id":234362843,"user":{"login":"phaas","id":1450137,"avatar_url":"https://avatars.githubusercontent.com/u/1450137?v=3","gravatar_id":"","url":"https://api.github.com/users/phaas","html_url":"https://github.com/phaas","followers_url":"https://api.github.com/users/phaas/followers","following_url":"https://api.github.com/users/phaas/following{/other_user}","gists_url":"https://api.github.com/users/phaas/gists{/gist_id}","starred_url":"https://api.github.com/users/phaas/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/phaas/subscriptions","organizations_url":"https://api.github.com/users/phaas/orgs","repos_url":"https://api.github.com/users/phaas/repos","events_url":"https://api.github.com/users/phaas/events{/privacy}","received_events_url":"https://api.github.com/users/phaas/received_events","type":"User","site_admin":false},"created_at":"2016-07-21T19:46:47Z","updated_at":"2016-07-21T19:46:47Z","body":"A workaround is to extend the serializer and perform the required encoding:\r\n\r\n```\r\n/**\r\n * Extension of KafkaAvroSerializer that supports reflection-based serialization of values when\r\n * objects are wrapped in a ReflectContainer.\r\n */\r\npublic class KafkaAvroReflectSerializer extends KafkaAvroSerializer {\r\n\tpublic static class ReflectContainer {\r\n\t\tprivate final Object object;\r\n\r\n\t\tpublic ReflectContainer(Object object) {\r\n\t\t\tthis.object = object;\r\n\t\t}\r\n\r\n\t\tpublic Object getObject() {\r\n\t\t\treturn object;\r\n\t\t}\r\n\t}\r\n\r\n\tprivate static final EncoderFactory ENCODER_FACTORY = EncoderFactory.get();\r\n\tprivate static final ReflectData REFLECT_DATA = ReflectData.get();\r\n\r\n\t@Override\r\n\tprotected byte[] serializeImpl(String subject, Object object) throws SerializationException {\r\n\t\tif (object instanceof ReflectContainer) {\r\n\t\t\tObject value = ((ReflectContainer) object).getObject();\r\n\t\t\tif (value == null) {\r\n\t\t\t\treturn null;\r\n\t\t\t}\r\n\t\t\tSchema schema = REFLECT_DATA.getSchema(value.getClass());\r\n\t\t\ttry {\r\n\t\t\t\tint registeredSchemaId = this.schemaRegistry.register(subject, schema);\r\n\r\n\t\t\t\tByteArrayOutputStream out = new ByteArrayOutputStream();\r\n\t\t\t\tout.write(0);\r\n\t\t\t\tout.write(ByteBuffer.allocate(4).putInt(registeredSchemaId).array());\r\n\r\n\t\t\t\tDatumWriter<Object> dw = new ReflectDatumWriter<>(schema);\r\n\t\t\t\tEncoder encoder = ENCODER_FACTORY.directBinaryEncoder(out, null);\r\n\t\t\t\tdw.write(value, encoder);\r\n\t\t\t\tencoder.flush();\r\n\t\t\t\treturn out.toByteArray();\r\n\t\t\t} catch (RuntimeException | IOException e) {\r\n\t\t\t\tthrow new SerializationException(\"Error serializing Avro message\", e);\r\n\t\t\t} catch (RestClientException e) {\r\n\t\t\t\tthrow new SerializationException(\"Error registering Avro schema: \" + schema, e);\r\n\t\t\t}\r\n\t\t}\r\n\t\treturn super.serializeImpl(subject, object);\r\n\t}\r\n}\r\n```"}}
{"actor":{"display_login":"phaas"}}