talfco

public  String persist(byte[] msg, String index, String type, String id) throws IOException {
      if (checkForAvroSingleObjectEncoding(msg)) {
          String jsonDoc = convertAvroBinaryToJSON(msg);
          esPersistencyManager.createUpdateDocument(index,type,jsonDoc,id);
          return jsonDoc;
      }
      else {
          logger.error("Received message wasn't Avro Single Object encoded");
          throw new IOException("Received message wasn't Avro Single Object encoded");
      }

talfco

public  String persist(byte[] msg, String index, String type, String id) throws IOException {
      if (checkForAvroSingleObjectEncoding(msg)) {
          String jsonDoc = convertAvroBinaryToJSON(msg);
          esPersistencyManager.createUpdateDocument(index,type,jsonDoc,id);
          return jsonDoc;
      }
      else {
          logger.error("Received message wasn't Avro Single Object encoded");
          throw new IOException("Received message wasn't Avro Single Object encoded");
      }

talfco

private  String convertAvroBinaryToJSON(byte[] msg) throws IOException {
    long msgFingerprint = getAvroFingerprint(msg);
    Schema mgsSchema = registry.getSchema(msgFingerprint);
    byte[] payload = extractPayload(msg);
    
    DatumReader<Object> reader = null;
    if (msgFingerprint == this.backendComponentSchemaFingerprint)
        reader = new GenericDatumReader<>(mgsSchema);
    else
        reader = new GenericDatumReader<>(mgsSchema,this.backendComponentSchema);

talfco

{ "namespace": "business.model.strategy",
  "type": "record",
  "doc" : "Schema for the Coporate Business Model Strategy Part",
  "name": "BusinessModelStrategy",
  "fields": [
    { "name": "avro_fingerprint", "type": "long"},
    { "name": "index_ipid", "type": "string"},
    { "name": "last_update_timestamp", "type": "long", "logical-type":"time-micros"},
    { "name": "last_update_loginid", "type": "string"},
    { "name": "client_ipid", "type": "string"},

talfco

public JsonObject readDocumentByIdAsObject(String index, String id) throws IOException{
   return esClient.execute(new Get.Builder(index, id).build()).getJsonObject();
}

talfco

public void createUpdateDocument(String index, String type, String docJson, String id) throws IOException {
   Index esIndex;
   if (id == null) {
       esIndex = new Index.Builder(docJson).index(index).type(type).build();
   } else {
       esIndex = new Index.Builder(docJson).index(index).type(type).id(id).build();
   }
   JestResult jestResult =  esClient.execute(esIndex);
   if(jestResult.isSucceeded()) {
       logger.info("Document persisted");

talfco

package net.cloudburo.avro.registry;

import com.google.gson.JsonObject;
import org.apache.avro.Schema;
import net.cloudburo.elasticsearch.ESPersistencyManager;

public class ElasticSearchSchemaRegistry extends SchemaRegistry {

    private static final String esIndex = "avroschema";

talfco

{ "namespace": "business.model.strategy",
  "type": "record",
  "doc" : "A simple Avro example schema",
  "name": "Person",
  "fields": [
    { "name": "personid", "type": "long", 
      "name": "lastname", "type": "string",
      "name": "surname", "type": [ "null", { "type": "string" }],"default": null}
  ]
}

talfco

<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>1.9.2</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals>
                <goal>schema</goal>
            </goals>

talfco

import org.apache.avro.Schema;
import org.apache.avro.SchemaNormalization;

public abstract class SchemaRegistry {

    public  abstract long registerSchema(Schema schema) throws IOException;
    public  abstract Schema getSchema(long fingerprint) throws IOException;

    public long getSchemaFingerprint(Schema schema) {
        return SchemaNormalization.parsingFingerprint64(schema);