diff --git a/lang/java/avro/pom.xml b/lang/java/avro/pom.xml
index e0484db050c..233208a87cb 100644
--- a/lang/java/avro/pom.xml
+++ b/lang/java/avro/pom.xml
@@ -169,6 +169,11 @@
       <artifactId>joda-time</artifactId>
       <optional>true</optional>
     </dependency>
+    <dependency>
+      <groupId>org.yaml</groupId>
+      <artifactId>snakeyaml</artifactId>
+      <version>1.16</version>
+    </dependency>
   </dependencies>
 
 </project>
diff --git a/lang/java/avro/src/main/java/org/apache/avro/Schema.java b/lang/java/avro/src/main/java/org/apache/avro/Schema.java
index 9a201ce45fa..66fa63f1eec 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/Schema.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/Schema.java
@@ -22,6 +22,8 @@
 import java.io.InputStream;
 import java.io.StringReader;
 import java.io.StringWriter;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Iterator;
@@ -43,6 +45,7 @@
 import org.codehaus.jackson.JsonGenerator;
 import org.codehaus.jackson.map.ObjectMapper;
 import org.codehaus.jackson.node.DoubleNode;
+import org.yaml.snakeyaml.Yaml;
 
 /** An abstract data type.
  * <p>A schema may be one of:
@@ -82,6 +85,7 @@ public abstract class Schema extends JsonProperties {
   static final ObjectMapper MAPPER = new ObjectMapper(FACTORY);
 
   private static final int NO_HASHCODE = Integer.MIN_VALUE;
+  public static String metadataDirectory = "";
 
   static {
     FACTORY.enable(JsonParser.Feature.ALLOW_COMMENTS);
@@ -283,6 +287,16 @@ public boolean isError() {
     throw new AvroRuntimeException("Not a record: "+this);
   }
 
+  /** If this is a record and there is a metadata return metadata */
+  public Map getMetadata() {
+    throw new AvroRuntimeException("Not a record: "+this);
+  }
+
+  /** If this is a record set metadata for this record. **/
+  public void setMetadata(Map metadata) {
+    throw new AvroRuntimeException("Not a record: "+this);
+  }
+
   /** If this is an array, returns its element type. */
   public Schema getElementType() {
     throw new AvroRuntimeException("Not an array: "+this);
@@ -605,6 +619,8 @@ private static class RecordSchema extends NamedSchema {
     private List<Field> fields;
     private Map<String, Field> fieldMap;
     private final boolean isError;
+    private Map metadata;
+
     public RecordSchema(Name name, String doc, boolean isError) {
       super(Type.RECORD, name, doc);
       this.isError = isError;
@@ -656,6 +672,18 @@ public void setFields(List<Field> fields) {
       this.fields = ff.lock();
       this.hashCode = NO_HASHCODE;
     }
+
+    @Override
+    public Map getMetadata() {
+      return metadata;
+    }
+
+    @Override
+    public void setMetadata(Map metadata) {
+      this.metadata = metadata;
+
+    }
+
     public boolean equals(Object o) {
       if (o == this) return true;
       if (!(o instanceof RecordSchema)) return false;
@@ -1249,6 +1277,7 @@ static Schema parse(JsonNode schema, Names names) {
       } else if (type.equals("record") || type.equals("error")) { // record
         List<Field> fields = new ArrayList<Field>();
         result = new RecordSchema(name, doc, type.equals("error"));
+        result.setMetadata(getMetadataFromRecord(name.name));
         if (name != null) names.add(result);
         JsonNode fieldsNode = schema.get("fields");
         if (fieldsNode == null || !fieldsNode.isArray())
@@ -1343,6 +1372,25 @@ static Schema parse(JsonNode schema, Names names) {
     }
   }
 
+  private static Map getMetadataFromRecord(String fileName) {
+    Map map = null;
+    String nameWithExtension = fileName + ".yml";
+    File metadataFileFromRecord = new File(Schema.metadataDirectory + "/" + nameWithExtension);
+
+    if(metadataFileFromRecord.exists()) {
+      try {
+        FileInputStream metadataInputStream = new FileInputStream(metadataFileFromRecord);
+        Yaml yaml = new Yaml();
+        map = (Map) yaml.load(metadataInputStream);
+
+      } catch (FileNotFoundException e) {
+        throw new RuntimeException(e);
+      }
+    }
+
+    return map;
+  }
+
   private static Set<String> parseAliases(JsonNode node) {
     JsonNode aliasesNode = node.get("aliases");
     if (aliasesNode == null)
diff --git a/lang/java/avro/src/test/java/org/apache/avro/TestSchema.java b/lang/java/avro/src/test/java/org/apache/avro/TestSchema.java
index ba2cab49d72..17c2cfc3f75 100644
--- a/lang/java/avro/src/test/java/org/apache/avro/TestSchema.java
+++ b/lang/java/avro/src/test/java/org/apache/avro/TestSchema.java
@@ -22,6 +22,8 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
+import java.io.File;
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -99,5 +101,14 @@ public void testSchemaWithNullFields() {
     Schema.createRecord("foobar", null, null, false, null);
   }
 
+  @Test
+  public void testRecordMetadadaLoad() throws IOException {
+    Schema.metadataDirectory = "src/test/resources/record-metadata";
+    File file = new File("src/test/resources/SchemaBuilder.avsc");
+    Schema schema = new Schema.Parser().parse(file);
+
+    assertNotNull(schema);
+    assertNotNull(schema.getMetadata());
+  }
 
 }
diff --git a/lang/java/avro/src/test/resources/record-metadata/recordAll.yml b/lang/java/avro/src/test/resources/record-metadata/recordAll.yml
new file mode 100644
index 00000000000..7b73a9f2a66
--- /dev/null
+++ b/lang/java/avro/src/test/resources/record-metadata/recordAll.yml
@@ -0,0 +1,3 @@
+persistence:
+  db:
+    database: 'foobar'
diff --git a/lang/java/maven-plugin/src/main/java/org/apache/avro/mojo/AbstractAvroMojo.java b/lang/java/maven-plugin/src/main/java/org/apache/avro/mojo/AbstractAvroMojo.java
index a5d8b311f11..278011eea01 100644
--- a/lang/java/maven-plugin/src/main/java/org/apache/avro/mojo/AbstractAvroMojo.java
+++ b/lang/java/maven-plugin/src/main/java/org/apache/avro/mojo/AbstractAvroMojo.java
@@ -18,18 +18,17 @@
 
 package org.apache.avro.mojo;
 
-import java.io.File;
-import java.io.IOException;
-import java.util.Arrays;
-
 import org.apache.avro.compiler.specific.SpecificCompiler;
-
 import org.apache.maven.plugin.AbstractMojo;
 import org.apache.maven.plugin.MojoExecutionException;
 import org.apache.maven.project.MavenProject;
 import org.apache.maven.shared.model.fileset.FileSet;
 import org.apache.maven.shared.model.fileset.util.FileSetManager;
 
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+
 /**
  * Base for Avro Compiler Mojos.
  */
@@ -114,6 +113,13 @@ public abstract class AbstractAvroMojo extends AbstractMojo {
    */
   protected String templateDirectory = "/org/apache/avro/compiler/specific/templates/java/classic/";
 
+  /**
+   * The directory that contains yaml files to be used as Metadata attribute in RecordSchema.
+   *
+   * @parameter property="metadataDirectory"
+   */
+  protected String metadataDirectory = "";
+
   /**
    * Determines whether or not to create setters for the fields of the record.
    * The default is to create setters.
diff --git a/lang/java/maven-plugin/src/main/java/org/apache/avro/mojo/SchemaMojo.java b/lang/java/maven-plugin/src/main/java/org/apache/avro/mojo/SchemaMojo.java
index 7a7eaf96aab..6f8c092722f 100644
--- a/lang/java/maven-plugin/src/main/java/org/apache/avro/mojo/SchemaMojo.java
+++ b/lang/java/maven-plugin/src/main/java/org/apache/avro/mojo/SchemaMojo.java
@@ -18,14 +18,13 @@
 
 package org.apache.avro.mojo;
 
+import org.apache.avro.Schema;
+import org.apache.avro.compiler.specific.SpecificCompiler;
 import org.apache.avro.generic.GenericData.StringType;
 
 import java.io.File;
 import java.io.IOException;
 
-import org.apache.avro.Schema;
-import org.apache.avro.compiler.specific.SpecificCompiler;
-
 /**
  * Generate Java classes from Avro schema files (.avsc)
  * 
@@ -61,18 +60,8 @@ public class SchemaMojo extends AbstractAvroMojo {
   @Override
   protected void doCompile(String filename, File sourceDirectory, File outputDirectory) throws IOException {
     File src = new File(sourceDirectory, filename);
-    Schema schema;
+    Schema schema = parseSrc(src);
 
-    // This is necessary to maintain backward-compatibility. If there are  
-    // no imported files then isolate the schemas from each other, otherwise
-    // allow them to share a single schema so resuse and sharing of schema
-    // is possible.
-    if (imports == null) {
-      schema = new Schema.Parser().parse(src);
-    } else {
-      schema = schemaParser.parse(src);
-    }
-    
     SpecificCompiler compiler = new SpecificCompiler(schema);
     compiler.setTemplateDir(templateDirectory);
     compiler.setStringType(StringType.valueOf(stringType));
@@ -91,4 +80,21 @@ protected String[] getIncludes() {
   protected String[] getTestIncludes() {
     return testIncludes;
   }
+
+  private Schema parseSrc(File src) throws IOException {
+    Schema.metadataDirectory = this.metadataDirectory;
+    Schema schema;
+
+    // This is necessary to maintain backward-compatibility. If there are
+    // no imported files then isolate the schemas from each other, otherwise
+    // allow them to share a single schema so reuse and sharing of schema
+    // is possible.
+    if (imports == null) {
+      schema = new Schema.Parser().parse(src);
+    } else {
+      schema = schemaParser.parse(src);
+    }
+
+    return schema;
+  }
 }