improve relay performance

databus-core/databus-core-schemas/src/main/java/com/linkedin/databus2/schemas/VersionedSchema.java

@@ -35,10 +35,13 @@ public class VersionedSchema
   private final String _origSchemaStr;
   private final List<Schema.Field> _pkFieldList;
 
+  private final SchemaId _schemaId;
+
   public VersionedSchema(VersionedSchemaId id, Schema s, String origSchemaStr)
   {
     _schema = s;
     _id = id;
+    _schemaId = SchemaId.createWithMd5(s);
     _origSchemaStr = origSchemaStr;
     _pkFieldList = new ArrayList<Schema.Field>();
   }
@@ -61,7 +64,7 @@ public Schema getSchema()
   @Override
   public String toString()
   {
-    return "(" + getSchemaBaseName() + ","  + getVersion() + "," + _schema + ")";
+    return "(" + getSchemaBaseName() + ","  + getVersion() + "," + _schema + ")," + _schemaId + ")";
   }
 
   /**
@@ -90,12 +93,16 @@ public VersionedSchemaId getId()
   {
     return _id;
   }
-  
+
   public List<Schema.Field> getPkFieldList()
   {
 	return _pkFieldList;
   }
 
+  public SchemaId getSchemaId() {
+    return _schemaId;
+  }
+
   /**
    * @return The original schema string as registered. Returns null if the original schema string is not available.
    */

databus2-relay/databus2-event-producer-common/src/main/java/com/linkedin/databus2/producers/ds/DbChangeEntry.java

@@ -1,5 +1,6 @@
 package com.linkedin.databus2.producers.ds;
 
+import com.linkedin.databus2.schemas.SchemaId;
 import java.util.List;
 
 import org.apache.avro.Schema;
@@ -63,6 +64,12 @@ public class DbChangeEntry {
 	 */
 	private final Schema _schema;
 
+	/**
+	 * Avro schema md5 id.
+	 */
+	private final SchemaId _schemaId;
+
+
 	/**
 	 * Primary Key(s) corresponding to the entry
 	 */
@@ -88,6 +95,10 @@ public Schema getSchema() {
 		return _schema;
 	}
 
+	public SchemaId getSchemaId() {
+		return _schemaId;
+	}
+
 	public List<KeyPair> getPkeys() {
 		return _pkeys;
 	}
@@ -125,14 +136,15 @@ public boolean equals(Object obj) {
 	}
 
 	public DbChangeEntry(long scn, long timestampNanos, GenericRecord record, DbusOpcode opCode,
-			boolean isReplicated, Schema schema, List<KeyPair> pkeys) {
+			boolean isReplicated, Schema schema, SchemaId schemaId, List<KeyPair> pkeys) {
 		super();
 		this._scn = scn;
 		this._timestampInNanos = timestampNanos;
 		this._record = record;
 		this._opCode = opCode;
 		this._isReplicated = isReplicated;
 		this._schema = schema;
+		this._schemaId = schemaId;
 		this._pkeys = pkeys;
 	}
 }

databus2-relay/databus2-event-producer-or/src/main/java/com/linkedin/databus2/producers/ORListener.java

@@ -365,7 +365,7 @@ private void frameAvroRecord(long tableId, BinlogEventV4Header bh, List<Row> rl,
 
         List<KeyPair> kps = generateKeyPair(gr, vs);
 
-        DbChangeEntry db = new DbChangeEntry(scn, timestampInNanos, gr, doc, isReplicated, schema, kps);
+        DbChangeEntry db = new DbChangeEntry(scn, timestampInNanos, gr, doc, isReplicated, schema, vs.getSchemaId(), kps);
         _transaction.getPerSourceTransaction(_tableUriToSrcIdMap.get(tableName)).mergeDbChangeEntrySet(db);
       }
     } catch (NoSuchSchemaException ne)

databus2-relay/databus2-event-producer-or/src/main/java/com/linkedin/databus2/producers/OpenReplicatorAvroEventFactory.java

@@ -103,7 +103,7 @@ public int createAndAppendEvent(DbChangeEntry changeEntry,
     short lPartitionId = _partitionFunction.getPartition(eventKey);
 
     //Get the md5 for the schema
-    SchemaId schemaId = SchemaId.createWithMd5(changeEntry.getSchema());
+    SchemaId schemaId = changeEntry.getSchemaId();
 
     byte[] payload = serializeEvent(changeEntry.getRecord());