diff --git a/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/util/DataTypeUtils.java b/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/util/DataTypeUtils.java
index c6dcb261f6..3411b13a31 100644
--- a/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/util/DataTypeUtils.java
+++ b/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/util/DataTypeUtils.java
@@ -1080,15 +1080,11 @@ public class DataTypeUtils {
         if (value instanceof BigInteger) {
             return (BigInteger) value;
         }
-        if (value instanceof Long) {
-            return BigInteger.valueOf((Long) value);
-        }
-        if (value instanceof Integer) {
-            return BigInteger.valueOf(((Integer) value).longValue());
-        }
-        if (value instanceof Short) {
-            return BigInteger.valueOf(((Short) value).longValue());
+
+        if (value instanceof Number) {
+            return BigInteger.valueOf(((Number) value).longValue());
         }
+
         if (value instanceof String) {
             try {
                 return new BigInteger((String) value);
@@ -1102,11 +1098,7 @@ public class DataTypeUtils {
     }
 
     public static boolean isBigIntTypeCompatible(final Object value) {
-        return value instanceof BigInteger
-                || value instanceof Long
-                || value instanceof Integer
-                || value instanceof Short
-                || value instanceof String;
+        return isNumberTypeCompatible(value, DataTypeUtils::isIntegral);
     }
 
     public static Boolean toBoolean(final Object value, final String fieldName) {
@@ -1277,7 +1269,10 @@ public class DataTypeUtils {
         return false;
     }
 
-    private static boolean isIntegral(final String value, final long minValue, final long maxValue) {
+    /**
+     * Check if the value is an integral.
+     */
+    private static boolean isIntegral(final String value) {
         if (value == null || value.isEmpty()) {
             return false;
         }
@@ -1298,6 +1293,18 @@ public class DataTypeUtils {
             }
         }
 
+        return true;
+    }
+
+    /**
+     * Check if the value is an integral within a value range.
+     */
+    private static boolean isIntegral(final String value, final long minValue, final long maxValue) {
+
+        if (!isIntegral(value)) {
+            return false;
+        }
+
         try {
             final long longValue = Long.parseLong(value);
             return longValue >= minValue && longValue <= maxValue;
@@ -1307,7 +1314,6 @@ public class DataTypeUtils {
         }
     }
 
-
     public static Integer toInteger(final Object value, final String fieldName) {
         if (value == null) {
             return null;
diff --git a/nifi-commons/nifi-record/src/test/java/org/apache/nifi/serialization/record/TestDataTypeUtils.java b/nifi-commons/nifi-record/src/test/java/org/apache/nifi/serialization/record/TestDataTypeUtils.java
index c64ed376d7..89a0490f58 100644
--- a/nifi-commons/nifi-record/src/test/java/org/apache/nifi/serialization/record/TestDataTypeUtils.java
+++ b/nifi-commons/nifi-record/src/test/java/org/apache/nifi/serialization/record/TestDataTypeUtils.java
@@ -30,6 +30,7 @@ import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Function;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -292,27 +293,32 @@ public class TestDataTypeUtils {
 
     @Test
     public void testIsCompatibleDataTypeBigint() {
-        assertTrue(DataTypeUtils.isCompatibleDataType(new BigInteger("12345678901234567890"), RecordFieldType.BIGINT.getDataType()));
-        assertTrue(DataTypeUtils.isCompatibleDataType(1234567890123456789L, RecordFieldType.BIGINT.getDataType()));
-        assertTrue(DataTypeUtils.isCompatibleDataType(1, RecordFieldType.BIGINT.getDataType()));
-        assertTrue(DataTypeUtils.isCompatibleDataType((short) 1, RecordFieldType.BIGINT.getDataType()));
-        assertTrue(DataTypeUtils.isCompatibleDataType("12345678901234567890", RecordFieldType.BIGINT.getDataType()));
-        assertTrue(DataTypeUtils.isCompatibleDataType("1234567XYZ", RecordFieldType.BIGINT.getDataType())); // Compatible but the value might not be a valid BigInteger
-        assertFalse(DataTypeUtils.isCompatibleDataType(3.0f, RecordFieldType.BIGINT.getDataType()));
-        assertFalse(DataTypeUtils.isCompatibleDataType(3.0, RecordFieldType.BIGINT.getDataType()));
-        assertFalse(DataTypeUtils.isCompatibleDataType(new Long[]{1L, 2L}, RecordFieldType.BIGINT.getDataType()));
+        final DataType dataType = RecordFieldType.BIGINT.getDataType();
+        assertTrue(DataTypeUtils.isCompatibleDataType(new BigInteger("12345678901234567890"), dataType));
+        assertTrue(DataTypeUtils.isCompatibleDataType(1234567890123456789L, dataType));
+        assertTrue(DataTypeUtils.isCompatibleDataType(1, dataType));
+        assertTrue(DataTypeUtils.isCompatibleDataType((short) 1, dataType));
+        assertTrue(DataTypeUtils.isCompatibleDataType("12345678901234567890", dataType));
+        assertTrue(DataTypeUtils.isCompatibleDataType(3.1f, dataType));
+        assertTrue(DataTypeUtils.isCompatibleDataType(3.0, dataType));
+        assertFalse(DataTypeUtils.isCompatibleDataType("1234567XYZ", dataType));
+        assertFalse(DataTypeUtils.isCompatibleDataType(new Long[]{1L, 2L}, dataType));
     }
 
     @Test
     public void testConvertDataTypeBigint() {
-        assertTrue(DataTypeUtils.convertType(new BigInteger("12345678901234567890"), RecordFieldType.BIGINT.getDataType(), "field") instanceof BigInteger);
-        assertTrue(DataTypeUtils.convertType(1234567890123456789L, RecordFieldType.BIGINT.getDataType(), "field") instanceof BigInteger);
-        assertTrue(DataTypeUtils.convertType(1, RecordFieldType.BIGINT.getDataType(), "field") instanceof BigInteger);
-        assertTrue(DataTypeUtils.convertType((short) 1, RecordFieldType.BIGINT.getDataType(), "field") instanceof BigInteger);
-        assertTrue(DataTypeUtils.convertType("12345678901234567890", RecordFieldType.BIGINT.getDataType(), "field") instanceof BigInteger);
+        final Function<Object, BigInteger> toBigInteger = v -> (BigInteger) DataTypeUtils.convertType(v, RecordFieldType.BIGINT.getDataType(), "field");
+        assertEquals(new BigInteger("12345678901234567890"), toBigInteger.apply(new BigInteger("12345678901234567890")));
+        assertEquals(new BigInteger("1234567890123456789"), toBigInteger.apply(1234567890123456789L));
+        assertEquals(new BigInteger("1"), toBigInteger.apply(1));
+        assertEquals(new BigInteger("1"), toBigInteger.apply((short) 1));
+        // Decimals are truncated.
+        assertEquals(new BigInteger("3"), toBigInteger.apply(3.4f));
+        assertEquals(new BigInteger("3"), toBigInteger.apply(3.9f));
+        assertEquals(new BigInteger("12345678901234567890"), toBigInteger.apply("12345678901234567890"));
         Exception e = null;
         try {
-            DataTypeUtils.convertType("1234567XYZ", RecordFieldType.BIGINT.getDataType(), "field");
+            toBigInteger.apply("1234567XYZ");
         } catch (IllegalTypeConversionException itce) {
             e = itce;
         }