diff --git a/lang/java/mapred/src/main/java/org/apache/avro/hadoop/io/AvroDatumConverterFactory.java b/lang/java/mapred/src/main/java/org/apache/avro/hadoop/io/AvroDatumConverterFactory.java
index f2e4fcca3d7..a9984e08e54 100644
--- a/lang/java/mapred/src/main/java/org/apache/avro/hadoop/io/AvroDatumConverterFactory.java
+++ b/lang/java/mapred/src/main/java/org/apache/avro/hadoop/io/AvroDatumConverterFactory.java
@@ -199,7 +199,7 @@ public BytesWritableConverter() {
     /** {@inheritDoc} */
     @Override
     public ByteBuffer convert(BytesWritable input) {
-      return ByteBuffer.wrap(input.getBytes());
+      return ByteBuffer.wrap(input.getBytes(), 0, input.getLength());
     }
 
     /** {@inheritDoc} */
diff --git a/lang/java/mapred/src/test/java/org/apache/avro/hadoop/io/TestAvroDatumConverterFactory.java b/lang/java/mapred/src/test/java/org/apache/avro/hadoop/io/TestAvroDatumConverterFactory.java
index b0d2041c1a2..21a1bf361d7 100644
--- a/lang/java/mapred/src/test/java/org/apache/avro/hadoop/io/TestAvroDatumConverterFactory.java
+++ b/lang/java/mapred/src/test/java/org/apache/avro/hadoop/io/TestAvroDatumConverterFactory.java
@@ -89,6 +89,21 @@ void convertBytesWritable() {
     assertEquals(3, bytes.get(2));
   }
 
+  @Test
+  void convertBytesWritableRespectsLogicalLength() {
+    AvroDatumConverter<BytesWritable, ByteBuffer> converter = mFactory.create(BytesWritable.class);
+    byte[] backing = new byte[] { 1, 2, 3, 4, 5 };
+    BytesWritable writable = new BytesWritable(backing);
+    writable.setSize(3);
+
+    ByteBuffer bytes = converter.convert(writable);
+
+    assertEquals(3, bytes.remaining());
+    assertEquals(1, bytes.get(0));
+    assertEquals(2, bytes.get(1));
+    assertEquals(3, bytes.get(2));
+  }
+
   @Test
   void convertByteWritable() {
     AvroDatumConverter<ByteWritable, GenericFixed> converter = mFactory.create(ByteWritable.class);