From ba513447d75dc5e95ddcdfcac1a1fefe2eb175ce Mon Sep 17 00:00:00 2001
From: Matt Burgess <mattyb149@apache.org>
Date: Mon, 21 Nov 2016 16:33:12 -0500
Subject: [PATCH] NIFI-3076: Fixed handling of 'medium' unsigned integers in
 JdbcCommon

This closes #1254.

Signed-off-by: Bryan Bende <bbende@apache.org>
---
 .../processors/standard/util/JdbcCommon.java  |  3 +-
 .../standard/util/TestJdbcCommon.java         | 36 +++++++++++++++++++
 2 files changed, 38 insertions(+), 1 deletion(-)

diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/JdbcCommon.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/JdbcCommon.java
index 22301b021e..49be067a8a 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/JdbcCommon.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/JdbcCommon.java
@@ -72,6 +72,7 @@ import org.apache.commons.lang3.StringUtils;
 public class JdbcCommon {
 
     private static final int MAX_DIGITS_IN_BIGINT = 19;
+    private static final int MAX_DIGITS_IN_INT = 9;
 
     public static long convertToAvroStream(final ResultSet rs, final OutputStream outStream, boolean convertNames) throws SQLException, IOException {
         return convertToAvroStream(rs, outStream, null, null, convertNames);
@@ -284,7 +285,7 @@ public class JdbcCommon {
                     break;
 
                 case INTEGER:
-                    if (meta.isSigned(i)) {
+                    if (meta.isSigned(i) || (meta.getPrecision(i) > 0 && meta.getPrecision(i) <= MAX_DIGITS_IN_INT)) {
                         builder.name(columnName).type().unionOf().nullBuilder().endNull().and().intType().endUnion().noDefault();
                     } else {
                         builder.name(columnName).type().unionOf().nullBuilder().endNull().and().longType().endUnion().noDefault();
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/TestJdbcCommon.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/TestJdbcCommon.java
index 18621d6417..f8ce1f3c01 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/TestJdbcCommon.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/TestJdbcCommon.java
@@ -282,6 +282,7 @@ public class TestJdbcCommon {
         final ResultSetMetaData metadata = mock(ResultSetMetaData.class);
         when(metadata.getColumnCount()).thenReturn(1);
         when(metadata.getColumnType(1)).thenReturn(Types.INTEGER);
+        when(metadata.getPrecision(1)).thenReturn(10);
         when(metadata.isSigned(1)).thenReturn(false);
         when(metadata.getColumnName(1)).thenReturn("Col1");
         when(metadata.getTableName(1)).thenReturn("Table1");
@@ -311,6 +312,41 @@ public class TestJdbcCommon {
         assertTrue(foundNullSchema);
     }
 
+    @Test
+    public void testMediumUnsignedIntShouldBeInt() throws SQLException, IllegalArgumentException, IllegalAccessException {
+        final ResultSetMetaData metadata = mock(ResultSetMetaData.class);
+        when(metadata.getColumnCount()).thenReturn(1);
+        when(metadata.getColumnType(1)).thenReturn(Types.INTEGER);
+        when(metadata.getPrecision(1)).thenReturn(8);
+        when(metadata.isSigned(1)).thenReturn(false);
+        when(metadata.getColumnName(1)).thenReturn("Col1");
+        when(metadata.getTableName(1)).thenReturn("Table1");
+
+        final ResultSet rs = mock(ResultSet.class);
+        when(rs.getMetaData()).thenReturn(metadata);
+
+        Schema schema = JdbcCommon.createSchema(rs);
+        Assert.assertNotNull(schema);
+
+        Schema.Field field = schema.getField("Col1");
+        Schema fieldSchema = field.schema();
+        Assert.assertEquals(2, fieldSchema.getTypes().size());
+
+        boolean foundIntSchema = false;
+        boolean foundNullSchema = false;
+
+        for (Schema type : fieldSchema.getTypes()) {
+            if (type.getType().equals(Schema.Type.INT)) {
+                foundIntSchema = true;
+            } else if (type.getType().equals(Schema.Type.NULL)) {
+                foundNullSchema = true;
+            }
+        }
+
+        assertTrue(foundIntSchema);
+        assertTrue(foundNullSchema);
+    }
+
 
     @Test
     public void testConvertToAvroStreamForBigDecimal() throws SQLException, IOException {