diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericData.java b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericData.java
index 2f2a103eb07..3a7505a04aa 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericData.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericData.java
@@ -317,30 +317,16 @@ public String toString() {
     }
   }
 
-  /** Default implementation of an array. */
-  @SuppressWarnings(value = "unchecked")
-  public static class Array<T> extends AbstractList<T> implements GenericArray<T>, Comparable<GenericArray<T>> {
-    private static final Object[] EMPTY = new Object[0];
+  public static abstract class AbstractArray<T> extends AbstractList<T>
+      implements GenericArray<T>, Comparable<GenericArray<T>> {
     private final Schema schema;
-    private int size;
-    private Object[] elements = EMPTY;
 
-    public Array(int capacity, Schema schema) {
-      if (schema == null || !Type.ARRAY.equals(schema.getType()))
-        throw new AvroRuntimeException("Not an array schema: " + schema);
-      this.schema = schema;
-      if (capacity != 0)
-        elements = new Object[capacity];
-    }
+    protected int size = 0;
 
-    public Array(Schema schema, Collection<T> c) {
+    public AbstractArray(Schema schema) {
       if (schema == null || !Type.ARRAY.equals(schema.getType()))
         throw new AvroRuntimeException("Not an array schema: " + schema);
       this.schema = schema;
-      if (c != null) {
-        elements = new Object[c.size()];
-        addAll(c);
-      }
     }
 
     @Override
@@ -353,23 +339,14 @@ public int size() {
       return size;
     }
 
-    @Override
-    public void clear() {
-      // Let GC do its work
-      Arrays.fill(elements, 0, size, null);
-      size = 0;
-    }
-
     @Override
     public void reset() {
       size = 0;
     }
 
     @Override
-    public void prune() {
-      if (size < elements.length) {
-        Arrays.fill(elements, size, elements.length, null);
-      }
+    public int compareTo(GenericArray<T> that) {
+      return GenericData.get().compare(this, that, this.getSchema());
     }
 
     @Override
@@ -384,7 +361,7 @@ public boolean hasNext() {
 
         @Override
         public T next() {
-          return (T) elements[position++];
+          return AbstractArray.this.get(position++);
         }
 
         @Override
@@ -394,6 +371,57 @@ public void remove() {
       };
     }
 
+    @Override
+    public void reverse() {
+      int left = 0;
+      int right = size - 1;
+
+      while (left < right) {
+        this.swap(left, right);
+
+        left++;
+        right--;
+      }
+    }
+
+    protected abstract void swap(int index1, int index2);
+  }
+
+  /** Default implementation of an array. */
+  @SuppressWarnings(value = "unchecked")
+  public static class Array<T> extends AbstractArray<T> {
+    private static final Object[] EMPTY = new Object[0];
+
+    private Object[] elements = EMPTY;
+
+    public Array(int capacity, Schema schema) {
+      super(schema);
+      if (capacity != 0)
+        elements = new Object[capacity];
+    }
+
+    public Array(Schema schema, Collection<T> c) {
+      super(schema);
+      if (c != null) {
+        elements = new Object[c.size()];
+        addAll(c);
+      }
+    }
+
+    @Override
+    public void clear() {
+      // Let GC do its work
+      Arrays.fill(elements, 0, size, null);
+      size = 0;
+    }
+
+    @Override
+    public void prune() {
+      if (size < elements.length) {
+        Arrays.fill(elements, size, elements.length, null);
+      }
+    }
+
     @Override
     public T get(int i) {
       if (i >= size)
@@ -447,18 +475,10 @@ public int compareTo(GenericArray<T> that) {
     }
 
     @Override
-    public void reverse() {
-      int left = 0;
-      int right = elements.length - 1;
-
-      while (left < right) {
-        Object tmp = elements[left];
-        elements[left] = elements[right];
-        elements[right] = tmp;
-
-        left++;
-        right--;
-      }
+    protected void swap(final int index1, final int index2) {
+      Object tmp = elements[index1];
+      elements[index1] = elements[index2];
+      elements[index2] = tmp;
     }
   }
 
@@ -1499,8 +1519,24 @@ public Object newArray(Object old, int size, Schema schema) {
     } else if (old instanceof Collection) {
       ((Collection<?>) old).clear();
       return old;
-    } else
+    } else {
+      if (schema.getElementType().getType() == Type.INT) {
+        return new PrimitivesArrays.IntArray(size, schema);
+      }
+      if (schema.getElementType().getType() == Type.BOOLEAN) {
+        return new PrimitivesArrays.BooleanArray(size, schema);
+      }
+      if (schema.getElementType().getType() == Type.LONG) {
+        return new PrimitivesArrays.LongArray(size, schema);
+      }
+      if (schema.getElementType().getType() == Type.FLOAT) {
+        return new PrimitivesArrays.FloatArray(size, schema);
+      }
+      if (schema.getElementType().getType() == Type.DOUBLE) {
+        return new PrimitivesArrays.DoubleArray(size, schema);
+      }
       return new GenericData.Array<Object>(size, schema);
+    }
   }
 
   /**
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/PrimitivesArrays.java b/lang/java/avro/src/main/java/org/apache/avro/generic/PrimitivesArrays.java
new file mode 100644
index 00000000000..52e304ffe06
--- /dev/null
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/PrimitivesArrays.java
@@ -0,0 +1,584 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.avro.generic;
+
+import org.apache.avro.AvroRuntimeException;
+import org.apache.avro.Schema;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+public class PrimitivesArrays {
+
+  public static class IntArray extends GenericData.AbstractArray<Integer> {
+    private static final int[] EMPTY = new int[0];
+
+    private int[] elements = EMPTY;
+
+    public IntArray(int capacity, Schema schema) {
+      super(schema);
+      if (!Schema.Type.INT.equals(schema.getElementType().getType()))
+        throw new AvroRuntimeException("Not a int array schema: " + schema);
+      if (capacity != 0)
+        elements = new int[capacity];
+    }
+
+    public IntArray(Schema schema, Collection<Integer> c) {
+      super(schema);
+      if (c != null) {
+        elements = new int[c.size()];
+        addAll(c);
+      }
+    }
+
+    @Override
+    public void clear() {
+      size = 0;
+    }
+
+    @Override
+    public Integer get(int i) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      return elements[i];
+    }
+
+    @Override
+    public void add(int location, Integer o) {
+      if (o == null) {
+        return;
+      }
+      this.add(location, o.intValue());
+    }
+
+    public void add(int location, int o) {
+      if (location > size || location < 0) {
+        throw new IndexOutOfBoundsException("Index " + location + " out of bounds.");
+      }
+      if (size == elements.length) {
+        // Increase size by 1.5x + 1
+        final int newSize = size + (size >> 1) + 1;
+        elements = Arrays.copyOf(elements, newSize);
+      }
+      System.arraycopy(elements, location, elements, location + 1, size - location);
+      elements[location] = o;
+      size++;
+    }
+
+    @Override
+    public Integer set(int i, Integer o) {
+      if (o == null) {
+        return null;
+      }
+      return this.set(i, o.intValue());
+    }
+
+    public int set(int i, int o) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      int response = elements[i];
+      elements[i] = o;
+      return response;
+    }
+
+    @Override
+    public Integer remove(int i) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      int result = elements[i];
+      --size;
+      System.arraycopy(elements, i + 1, elements, i, (size - i));
+      return result;
+    }
+
+    @Override
+    public Integer peek() {
+      return (size < elements.length) ? elements[size] : null;
+    }
+
+    @Override
+    public int compareTo(GenericArray<Integer> that) {
+      return GenericData.get().compare(this, that, this.getSchema());
+    }
+
+    @Override
+    protected void swap(final int index1, final int index2) {
+      int tmp = elements[index1];
+      elements[index1] = elements[index2];
+      elements[index2] = tmp;
+    }
+  }
+
+  public static class LongArray extends GenericData.AbstractArray<Long> {
+    private static final long[] EMPTY = new long[0];
+
+    private long[] elements = EMPTY;
+
+    public LongArray(int capacity, Schema schema) {
+      super(schema);
+      if (!Schema.Type.LONG.equals(schema.getElementType().getType()))
+        throw new AvroRuntimeException("Not a long array schema: " + schema);
+      if (capacity != 0)
+        elements = new long[capacity];
+    }
+
+    public LongArray(Schema schema, Collection<Long> c) {
+      super(schema);
+      if (c != null) {
+        elements = new long[c.size()];
+        addAll(c);
+      }
+    }
+
+    @Override
+    public void clear() {
+      size = 0;
+    }
+
+    @Override
+    public Long get(int i) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      return elements[i];
+    }
+
+    @Override
+    public void add(int location, Long o) {
+      if (o == null) {
+        return;
+      }
+      this.add(location, o.longValue());
+    }
+
+    public void add(int location, long o) {
+      if (location > size || location < 0) {
+        throw new IndexOutOfBoundsException("Index " + location + " out of bounds.");
+      }
+      if (size == elements.length) {
+        // Increase size by 1.5x + 1
+        final int newSize = size + (size >> 1) + 1;
+        elements = Arrays.copyOf(elements, newSize);
+      }
+      System.arraycopy(elements, location, elements, location + 1, size - location);
+      elements[location] = o;
+      size++;
+    }
+
+    @Override
+    public Long set(int i, Long o) {
+      if (o == null) {
+        return null;
+      }
+      return this.set(i, o.longValue());
+    }
+
+    public long set(int i, long o) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      long response = elements[i];
+      elements[i] = o;
+      return response;
+    }
+
+    @Override
+    public Long remove(int i) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      long result = elements[i];
+      --size;
+      System.arraycopy(elements, i + 1, elements, i, (size - i));
+      return result;
+    }
+
+    @Override
+    public Long peek() {
+      return (size < elements.length) ? elements[size] : null;
+    }
+
+    @Override
+    public int compareTo(GenericArray<Long> that) {
+      return GenericData.get().compare(this, that, this.getSchema());
+    }
+
+    @Override
+    protected void swap(final int index1, final int index2) {
+      long tmp = elements[index1];
+      elements[index1] = elements[index2];
+      elements[index2] = tmp;
+    }
+  }
+
+  public static class BooleanArray extends GenericData.AbstractArray<Boolean> {
+    private static final byte[] EMPTY = new byte[0];
+
+    private byte[] elements = EMPTY;
+
+    public BooleanArray(int capacity, Schema schema) {
+      super(schema);
+      if (!Schema.Type.BOOLEAN.equals(schema.getElementType().getType()))
+        throw new AvroRuntimeException("Not a boolean array schema: " + schema);
+      if (capacity != 0)
+        elements = new byte[1 + (capacity / Byte.SIZE)];
+    }
+
+    public BooleanArray(Schema schema, Collection<Boolean> c) {
+      super(schema);
+
+      if (c != null) {
+        elements = new byte[1 + (c.size() / 8)];
+        if (c instanceof BooleanArray) {
+          BooleanArray other = (BooleanArray) c;
+          this.size = other.size;
+          System.arraycopy(other.elements, 0, this.elements, 0, other.elements.length);
+        } else {
+          addAll(c);
+        }
+      }
+    }
+
+    @Override
+    public void clear() {
+      size = 0;
+    }
+
+    @Override
+    public Boolean get(int i) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      return (elements[i / 8] & (1 << (i % 8))) > 0;
+    }
+
+    @Override
+    public boolean add(final Boolean o) {
+      if (o == null) {
+        return false;
+      }
+      return this.add(o.booleanValue());
+    }
+
+    public boolean add(final boolean o) {
+      if (this.size == elements.length * 8) {
+        final int newLength = elements.length + (elements.length >> 1) + 1;
+        elements = Arrays.copyOf(elements, newLength);
+      }
+      this.size++;
+      this.set(this.size - 1, o);
+      return true;
+    }
+
+    @Override
+    public void add(int location, Boolean o) {
+      if (o == null) {
+        return;
+      }
+      this.add(location, o.booleanValue());
+    }
+
+    public void add(int location, boolean o) {
+      if (location > size || location < 0) {
+        throw new IndexOutOfBoundsException("Index " + location + " out of bounds.");
+      }
+      if (size == elements.length * 8) {
+        // Increase size by 1.5x + 1
+        final int newLength = elements.length + (elements.length >> 1) + 1;
+        elements = Arrays.copyOf(elements, newLength);
+      }
+      size++;
+      for (int index = this.size / 8; index > (location / 8); index--) {
+        elements[index] <<= 1;
+        if (index > 0 && (elements[index - 1] & (1 << Byte.SIZE)) > 0) {
+          elements[index] |= 1;
+        }
+      }
+      byte pos = (byte) (1 << (location % Byte.SIZE));
+      byte highbits = (byte) ~(pos + (pos - 1));
+      byte lowbits = (byte) (pos - 1);
+      byte currentHigh = (byte) ((elements[location / 8] & highbits) << 1);
+      byte currentLow = (byte) (elements[location / 8] & lowbits);
+      if (o) {
+        elements[location / 8] = (byte) (currentHigh | currentLow | pos);
+      } else {
+        elements[location / 8] = (byte) (currentHigh | currentLow);
+      }
+
+    }
+
+    @Override
+    public Boolean set(int i, Boolean o) {
+      if (o == null) {
+        return null;
+      }
+      return this.set(i, o.booleanValue());
+    }
+
+    public boolean set(int i, boolean o) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      boolean response = (elements[i / 8] & (1 << (i % 8))) > 0;
+      if (o) {
+        elements[i / 8] |= 1 << (i % 8);
+      } else {
+        elements[i / 8] &= 0xFF - (1 << (i % 8));
+      }
+      return response;
+    }
+
+    @Override
+    public Boolean remove(int i) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      boolean result = (elements[(i / 8)] & (1 << (i % 8))) > 0;
+      --size;
+
+      byte memo = 0;
+      if ((i / 8) + 1 < elements.length) {
+        memo = (byte) ((1 & (elements[(i / 8) + 1])) << 7);
+      }
+      for (int index = (i / 8) + 1; index <= (size / 8); index++) {
+        elements[index] = (byte) ((elements[index] & 0xff) >>> 1);
+        if (index + 1 < elements.length && (elements[index + 1] & 1) == 1) {
+          elements[index] |= 1 << (Byte.SIZE - 1);
+        }
+      }
+      // 87654321 => <memo>8764321
+      byte start = (byte) ((1 << ((i + 1) % 8)) - 1);
+      byte end = (byte) ~start;
+      elements[i / 8] = (byte) ((((start & 0xff) >>> 1) & elements[i / 8]) // 1234
+          | (end & (elements[i / 8] >> 1)) // 876
+          | memo);
+
+      return result;
+    }
+
+    @Override
+    public Boolean peek() {
+      return (size < elements.length * Byte.SIZE) ? (elements[(size / 8)] & (1 << (size % 8))) > 0 : null;
+    }
+
+    @Override
+    public int compareTo(GenericArray<Boolean> that) {
+      return GenericData.get().compare(this, that, this.getSchema());
+    }
+
+    @Override
+    protected void swap(final int index1, final int index2) {
+      boolean tmp = this.get(index1);
+      this.set(index1, this.get(index2));
+      this.set(index2, tmp);
+    }
+  }
+
+  public static class FloatArray extends GenericData.AbstractArray<Float> {
+    private static final float[] EMPTY = new float[0];
+
+    private float[] elements = EMPTY;
+
+    public FloatArray(int capacity, Schema schema) {
+      super(schema);
+      if (!Schema.Type.FLOAT.equals(schema.getElementType().getType()))
+        throw new AvroRuntimeException("Not a float array schema: " + schema);
+      if (capacity != 0)
+        elements = new float[capacity];
+    }
+
+    public FloatArray(Schema schema, Collection<Float> c) {
+      super(schema);
+      if (c != null) {
+        elements = new float[c.size()];
+        addAll(c);
+      }
+    }
+
+    @Override
+    public void clear() {
+      size = 0;
+    }
+
+    @Override
+    public Float get(int i) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      return elements[i];
+    }
+
+    @Override
+    public void add(int location, Float o) {
+      if (o == null) {
+        return;
+      }
+      this.add(location, o.floatValue());
+    }
+
+    public void add(int location, float o) {
+      if (location > size || location < 0) {
+        throw new IndexOutOfBoundsException("Index " + location + " out of bounds.");
+      }
+      if (size == elements.length) {
+        // Increase size by 1.5x + 1
+        final int newSize = size + (size >> 1) + 1;
+        elements = Arrays.copyOf(elements, newSize);
+      }
+      System.arraycopy(elements, location, elements, location + 1, size - location);
+      elements[location] = o;
+      size++;
+    }
+
+    @Override
+    public Float set(int i, Float o) {
+      if (o == null) {
+        return null;
+      }
+      return this.set(i, o.floatValue());
+    }
+
+    public float set(int i, float o) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      float response = elements[i];
+      elements[i] = o;
+      return response;
+    }
+
+    @Override
+    public Float remove(int i) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      float result = elements[i];
+      --size;
+      System.arraycopy(elements, i + 1, elements, i, (size - i));
+      return result;
+    }
+
+    @Override
+    public Float peek() {
+      return (size < elements.length) ? elements[size] : null;
+    }
+
+    @Override
+    public int compareTo(GenericArray<Float> that) {
+      return GenericData.get().compare(this, that, this.getSchema());
+    }
+
+    @Override
+    protected void swap(final int index1, final int index2) {
+      float tmp = this.get(index1);
+      this.set(index1, this.get(index2));
+      this.set(index2, tmp);
+    }
+  }
+
+  public static class DoubleArray extends GenericData.AbstractArray<Double> {
+    private static final double[] EMPTY = new double[0];
+
+    private double[] elements = EMPTY;
+
+    public DoubleArray(int capacity, Schema schema) {
+      super(schema);
+      if (!Schema.Type.DOUBLE.equals(schema.getElementType().getType()))
+        throw new AvroRuntimeException("Not a double array schema: " + schema);
+      if (capacity != 0)
+        elements = new double[capacity];
+    }
+
+    public DoubleArray(Schema schema, Collection<Double> c) {
+      super(schema);
+      if (c != null) {
+        elements = new double[c.size()];
+        addAll(c);
+      }
+    }
+
+    @Override
+    public void clear() {
+      size = 0;
+    }
+
+    @Override
+    public Double get(int i) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      return elements[i];
+    }
+
+    @Override
+    public void add(int location, Double o) {
+      if (o == null) {
+        return;
+      }
+      this.add(location, o.floatValue());
+    }
+
+    public void add(int location, double o) {
+      if (location > size || location < 0) {
+        throw new IndexOutOfBoundsException("Index " + location + " out of bounds.");
+      }
+      if (size == elements.length) {
+        // Increase size by 1.5x + 1
+        final int newSize = size + (size >> 1) + 1;
+        elements = Arrays.copyOf(elements, newSize);
+      }
+      System.arraycopy(elements, location, elements, location + 1, size - location);
+      elements[location] = o;
+      size++;
+    }
+
+    @Override
+    public Double set(int i, Double o) {
+      if (o == null) {
+        return null;
+      }
+      return this.set(i, o.floatValue());
+    }
+
+    public double set(int i, double o) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      double response = elements[i];
+      elements[i] = o;
+      return response;
+    }
+
+    @Override
+    public Double remove(int i) {
+      if (i >= size)
+        throw new IndexOutOfBoundsException("Index " + i + " out of bounds.");
+      double result = elements[i];
+      --size;
+      System.arraycopy(elements, i + 1, elements, i, (size - i));
+      return result;
+    }
+
+    @Override
+    public Double peek() {
+      return (size < elements.length) ? elements[size] : null;
+    }
+
+    @Override
+    public int compareTo(GenericArray<Double> that) {
+      return GenericData.get().compare(this, that, this.getSchema());
+    }
+
+    @Override
+    protected void swap(final int index1, final int index2) {
+      double tmp = this.get(index1);
+      this.set(index1, this.get(index2));
+      this.set(index2, tmp);
+    }
+  }
+
+}
diff --git a/lang/java/avro/src/test/java/org/apache/avro/generic/GenericDataArrayTest.java b/lang/java/avro/src/test/java/org/apache/avro/generic/GenericDataArrayTest.java
new file mode 100644
index 00000000000..a4ffebac02d
--- /dev/null
+++ b/lang/java/avro/src/test/java/org/apache/avro/generic/GenericDataArrayTest.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.avro.generic;
+
+import org.apache.avro.Schema;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class GenericDataArrayTest {
+
+  @Test
+  void test() {
+    GenericData.Array<String> array = new GenericData.Array<>(10,
+        Schema.createArray(Schema.create(Schema.Type.STRING)));
+    array.add("One");
+    array.add("Two");
+    array.add("Two");
+    array.add("Three");
+    array.add(4, "Four");
+    array.remove(1);
+    Assertions.assertEquals(4, array.size());
+    Assertions.assertEquals("One", array.get(0));
+    Assertions.assertEquals("Two", array.get(1));
+    Assertions.assertEquals("Three", array.get(2));
+    Assertions.assertEquals("Four", array.get(3));
+  }
+
+}
diff --git a/lang/java/avro/src/test/java/org/apache/avro/generic/PrimitivesArraysTest.java b/lang/java/avro/src/test/java/org/apache/avro/generic/PrimitivesArraysTest.java
new file mode 100644
index 00000000000..7d199bf92c8
--- /dev/null
+++ b/lang/java/avro/src/test/java/org/apache/avro/generic/PrimitivesArraysTest.java
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.avro.generic;
+
+import org.apache.avro.Schema;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class PrimitivesArraysTest {
+
+  @Test
+  void booleanArray() {
+    PrimitivesArrays.BooleanArray ba = new PrimitivesArrays.BooleanArray(4,
+        Schema.createArray(Schema.create(Schema.Type.BOOLEAN)));
+
+    Assertions.assertEquals(0, ba.size());
+    for (int i = 1; i < 100; i++) {
+      if (i % 3 == 0 || i % 5 == 0) {
+        ba.add(true);
+      } else {
+        ba.add(false);
+      }
+    }
+    Assertions.assertEquals(99, ba.size());
+    for (int i = 1; i < 100; i++) {
+      if (i % 3 == 0 || i % 5 == 0) {
+        Assertions.assertTrue(ba.get(i - 1), "Error for " + i);
+      } else {
+        Assertions.assertFalse(ba.get(i - 1), "Error for " + i);
+      }
+    }
+    Assertions.assertFalse(ba.remove(12));
+    Assertions.assertEquals(98, ba.size());
+    for (int i = 13; i < 99; i++) {
+      if ((i + 1) % 3 == 0 || (i + 1) % 5 == 0) {
+        Assertions.assertTrue(ba.get(i - 1), "After delete, Error for " + i);
+      } else {
+        Assertions.assertFalse(ba.get(i - 1), "After delete, Error for " + i);
+      }
+    }
+
+    ba.add(12, false);
+    Assertions.assertEquals(99, ba.size());
+    for (int i = 1; i < 100; i++) {
+      if (i % 3 == 0 || i % 5 == 0) {
+        Assertions.assertTrue(ba.get(i - 1), "Error for " + i);
+      } else {
+        Assertions.assertFalse(ba.get(i - 1), "Error for " + i);
+      }
+    }
+    Assertions.assertFalse(ba.remove(12));
+    ba.add(12, true);
+    for (int i = 1; i < 100; i++) {
+      if (i % 3 == 0 || i % 5 == 0 || i == 13) {
+        Assertions.assertTrue(ba.get(i - 1), "Error for " + i);
+      } else {
+        Assertions.assertFalse(ba.get(i - 1), "Error for " + i);
+      }
+    }
+    ba.add(99, true);
+    Assertions.assertTrue(ba.get(99), "Error for 99");
+    ba.remove(99);
+    ba.reverse();
+    for (int i = 1; i < 100; i++) {
+      if (i % 3 == 0 || i % 5 == 0 || i == 13) {
+        Assertions.assertTrue(ba.get(99 - i), "Error for " + i);
+      } else {
+        Assertions.assertFalse(ba.get(99 - i), "Error for " + i);
+      }
+    }
+  }
+
+  @Test
+  void booleanArrayIterator() {
+    PrimitivesArrays.BooleanArray ba = new PrimitivesArrays.BooleanArray(4,
+        Schema.createArray(Schema.create(Schema.Type.BOOLEAN)));
+    boolean[] model = new boolean[] { true, false, false, true, true, true, false, false, true, false, false };
+    for (boolean x : model) {
+      ba.add(x);
+    }
+    Assertions.assertEquals(model.length, ba.size());
+    int index = 0;
+    for (Boolean b : ba) {
+      Assertions.assertEquals(model[index], b);
+      index++;
+    }
+  }
+
+  @Test
+  void intArray() {
+    final PrimitivesArrays.IntArray intArray = new PrimitivesArrays.IntArray(4,
+        Schema.createArray(Schema.create(Schema.Type.INT)));
+    for (int i = 1; i <= 100; i++) {
+      intArray.add(i);
+    }
+    Assertions.assertEquals(100, intArray.size());
+    for (int i = 1; i <= 100; i++) {
+      Assertions.assertEquals(i, intArray.get(i - 1));
+    }
+
+    int expectedValue = 1;
+    for (Integer value : intArray) {
+      Assertions.assertEquals(expectedValue, value);
+      expectedValue++;
+    }
+
+    intArray.remove(40);
+    Assertions.assertEquals(99, intArray.size());
+    for (int i = 1; i <= 99; i++) {
+      if (i <= 40) {
+        Assertions.assertEquals(i, intArray.get(i - 1));
+      } else {
+        Assertions.assertEquals(i + 1, intArray.get(i - 1));
+      }
+    }
+    intArray.add(40, 41);
+    Assertions.assertEquals(100, intArray.size());
+    for (int i = 1; i <= 100; i++) {
+      Assertions.assertEquals(i, intArray.get(i - 1));
+    }
+    intArray.set(40, 25);
+    Assertions.assertEquals(25, intArray.get(40));
+
+    Assertions.assertEquals(0, intArray.peek());
+    intArray.set(40, 41);
+    intArray.reverse();
+    Assertions.assertEquals(100, intArray.size());
+    for (int i = 1; i <= 100; i++) {
+      Assertions.assertEquals(101 - i, intArray.get(i - 1));
+    }
+  }
+
+  @Test
+  void longArray() {
+    final PrimitivesArrays.LongArray longArray = new PrimitivesArrays.LongArray(4,
+        Schema.createArray(Schema.create(Schema.Type.LONG)));
+    for (long i = 1; i <= 100; i++) {
+      longArray.add(i);
+    }
+    Assertions.assertEquals(100l, longArray.size());
+    for (int i = 1; i <= 100; i++) {
+      Assertions.assertEquals(i, longArray.get(i - 1));
+    }
+
+    int expectedValue = 1;
+    for (Long value : longArray) {
+      Assertions.assertEquals(expectedValue, value);
+      expectedValue++;
+    }
+
+    longArray.remove(40);
+    Assertions.assertEquals(99, longArray.size());
+    for (int i = 1; i <= 99; i++) {
+      if (i <= 40) {
+        Assertions.assertEquals(i, longArray.get(i - 1));
+      } else {
+        Assertions.assertEquals(i + 1, longArray.get(i - 1));
+      }
+    }
+    longArray.add(40, 41);
+    Assertions.assertEquals(100, longArray.size());
+    for (int i = 1; i <= 100; i++) {
+      Assertions.assertEquals(i, longArray.get(i - 1));
+    }
+    longArray.set(40, 25);
+    Assertions.assertEquals(25, longArray.get(40));
+
+    Assertions.assertEquals(0, longArray.peek());
+    longArray.set(40, 41);
+    longArray.reverse();
+    Assertions.assertEquals(100, longArray.size());
+    for (int i = 1; i <= 100; i++) {
+      Assertions.assertEquals(101 - i, longArray.get(i - 1));
+    }
+  }
+
+  @Test
+  void floatArray() {
+    final PrimitivesArrays.FloatArray floatArray = new PrimitivesArrays.FloatArray(4,
+        Schema.createArray(Schema.create(Schema.Type.FLOAT)));
+    for (int i = 1; i <= 100; i++) {
+      floatArray.add(i * 3.3f);
+    }
+    Assertions.assertEquals(100, floatArray.size());
+    for (int i = 1; i <= 100; i++) {
+      Assertions.assertEquals(i * 3.3f, floatArray.get(i - 1));
+    }
+
+    float expectedValue = 1.0f;
+    for (Float value : floatArray) {
+      Assertions.assertEquals(expectedValue * 3.3f, value);
+      expectedValue++;
+    }
+
+    floatArray.remove(40);
+    Assertions.assertEquals(99, floatArray.size());
+    for (int i = 1; i <= 99; i++) {
+      if (i <= 40) {
+        Assertions.assertEquals(i * 3.3f, floatArray.get(i - 1));
+      } else {
+        Assertions.assertEquals((i + 1) * 3.3f, floatArray.get(i - 1));
+      }
+    }
+    floatArray.add(40, 41 * 3.3f);
+    Assertions.assertEquals(100, floatArray.size());
+    for (int i = 1; i <= 100; i++) {
+      Assertions.assertEquals(i * 3.3f, floatArray.get(i - 1));
+    }
+    floatArray.set(40, 25.2f);
+    Assertions.assertEquals(25.2f, floatArray.get(40));
+
+    Assertions.assertEquals(0.0f, floatArray.peek());
+    floatArray.set(40, 41 * 3.3f);
+    floatArray.reverse();
+    Assertions.assertEquals(100, floatArray.size());
+    for (int i = 1; i <= 100; i++) {
+      Assertions.assertEquals((101 - i) * 3.3f, floatArray.get(i - 1));
+    }
+  }
+
+  @Test
+  void doubleArray() {
+    final PrimitivesArrays.DoubleArray doubleArray = new PrimitivesArrays.DoubleArray(4,
+        Schema.createArray(Schema.create(Schema.Type.DOUBLE)));
+    for (int i = 1; i <= 100; i++) {
+      doubleArray.add(i * 3.0d);
+    }
+    Assertions.assertEquals(100, doubleArray.size());
+    for (int i = 1; i <= 100; i++) {
+      Assertions.assertEquals(i * 3.0d, doubleArray.get(i - 1));
+    }
+
+    double expectedValue = 1.0f;
+    for (Double value : doubleArray) {
+      Assertions.assertEquals(expectedValue * 3.0d, value);
+      expectedValue++;
+    }
+
+    doubleArray.remove(40);
+    Assertions.assertEquals(99, doubleArray.size());
+    for (int i = 1; i <= 99; i++) {
+      if (i <= 40) {
+        Assertions.assertEquals(i * 3.0d, doubleArray.get(i - 1));
+      } else {
+        Assertions.assertEquals((i + 1) * 3.0d, doubleArray.get(i - 1));
+      }
+    }
+    doubleArray.add(40, 41 * 3.0d);
+    Assertions.assertEquals(100, doubleArray.size());
+    for (int i = 1; i <= 100; i++) {
+      Assertions.assertEquals(i * 3.0d, doubleArray.get(i - 1));
+    }
+    doubleArray.set(40, 25.2d);
+    Assertions.assertEquals(25.2d, doubleArray.get(40));
+
+    Assertions.assertEquals(0.0d, doubleArray.peek());
+    doubleArray.set(40, 41 * 3.0d);
+    doubleArray.reverse();
+    Assertions.assertEquals(100, doubleArray.size());
+    for (int i = 1; i <= 100; i++) {
+      Assertions.assertEquals((101 - i) * 3.0d, doubleArray.get(i - 1));
+    }
+  }
+}