#fixed code
                              @Test
  public void shouldReturnThePrimaryKeysAtTheEndWhenOneFieldIsPK() {
    Schema schema = SchemaBuilder.struct().name("com.example.Person")
            .field("firstName", Schema.STRING_SCHEMA)
            .field("lastName", Schema.STRING_SCHEMA)
            .field("age", Schema.INT32_SCHEMA)
            .field("bool", Schema.BOOLEAN_SCHEMA)
            .field("short", Schema.INT16_SCHEMA)
            .field("byte", Schema.INT8_SCHEMA)
            .field("long", Schema.INT64_SCHEMA)
            .field("float", Schema.FLOAT32_SCHEMA)
            .field("double", Schema.FLOAT64_SCHEMA)
            .field("bytes", Schema.BYTES_SCHEMA)
            .field("threshold", Schema.OPTIONAL_FLOAT64_SCHEMA).build();

    short s = 1234;
    byte b = -32;
    long l = 12425436;
    float f = (float) 2356.3;
    double d = -2436546.56457;
    byte[] bs = new byte[]{-32, 124};

    Struct struct = new Struct(schema)
            .put("firstName", "Alex")
            .put("lastName", "Smith")
            .put("bool", true)
            .put("short", s)
            .put("byte", b)
            .put("long", l)
            .put("float", f)
            .put("double", d)
            .put("bytes", bs)
            .put("age", 30);

    Map<String, FieldAlias> mappings = new HashMap<>();
    mappings.put("long", new FieldAlias("long", true));

    FieldsMappings tm = new FieldsMappings("table", "topic", true, mappings);

    StructFieldsDataExtractor dataExtractor = new StructFieldsDataExtractor(tm);
    List<PreparedStatementBinder> binders = dataExtractor.get(struct,
            new SinkRecord("", 2, null, null, schema, struct, 2));

    HashMap<String, PreparedStatementBinder> map = new HashMap<>();
    LinkedList<PreparedStatementBinder> pkBinders = new LinkedList<>();
    for (PreparedStatementBinder p : binders) {
      if (p.isPrimaryKey()) {
        pkBinders.add(p);
      }
      map.put(p.getFieldName(), p);
    }

    assertTrue(!binders.isEmpty());
    assertEquals(map.size(), 10);

    assertTrue(Objects.equals(pkBinders.get(0).getFieldName(), "long"));

    assertTrue(map.containsKey("firstName"));
    assertTrue(map.get("firstName").getClass() == StringPreparedStatementBinder.class);

    assertTrue(map.containsKey("lastName"));
    assertTrue(map.get("lastName").getClass() == StringPreparedStatementBinder.class);

    assertTrue(map.containsKey("age"));
    assertTrue(map.get("age").getClass() == IntPreparedStatementBinder.class);

    assertTrue(map.get("long").getClass() == LongPreparedStatementBinder.class);
    assertEquals(((LongPreparedStatementBinder) map.get("long")).getValue(), l);

    assertTrue(map.get("short").getClass() == ShortPreparedStatementBinder.class);
    assertEquals(((ShortPreparedStatementBinder) map.get("short")).getValue(), s);

    assertTrue(map.get("byte").getClass() == BytePreparedStatementBinder.class);
    assertEquals(((BytePreparedStatementBinder) map.get("byte")).getValue(), b);

    assertTrue(map.get("float").getClass() == FloatPreparedStatementBinder.class);
    assertEquals(Float.compare(((FloatPreparedStatementBinder) map.get("float")).getValue(), f), 0);

    assertTrue(map.get("double").getClass() == DoublePreparedStatementBinder.class);
    assertEquals(Double.compare(((DoublePreparedStatementBinder) map.get("double")).getValue(), d), 0);

    assertTrue(map.get("bytes").getClass() == BytesPreparedStatementBinder.class);
    assertTrue(Arrays.equals(bs, ((BytesPreparedStatementBinder) map.get("bytes")).getValue()));
  }
#vulnerable code 
                              @Test
  public void shouldReturnThePrimaryKeysAtTheEndWhenOneFieldIsPK() {
    Schema schema = SchemaBuilder.struct().name("com.example.Person")
            .field("firstName", Schema.STRING_SCHEMA)
            .field("lastName", Schema.STRING_SCHEMA)
            .field("age", Schema.INT32_SCHEMA)
            .field("bool", Schema.BOOLEAN_SCHEMA)
            .field("short", Schema.INT16_SCHEMA)
            .field("byte", Schema.INT8_SCHEMA)
            .field("long", Schema.INT64_SCHEMA)
            .field("float", Schema.FLOAT32_SCHEMA)
            .field("double", Schema.FLOAT64_SCHEMA)
            .field("bytes", Schema.BYTES_SCHEMA)
            .field("threshold", Schema.OPTIONAL_FLOAT64_SCHEMA).build();

    short s = 1234;
    byte b = -32;
    long l = 12425436;
    float f = (float) 2356.3;
    double d = -2436546.56457;
    byte[] bs = new byte[]{-32, 124};

    Struct struct = new Struct(schema)
            .put("firstName", "Alex")
            .put("lastName", "Smith")
            .put("bool", true)
            .put("short", s)
            .put("byte", b)
            .put("long", l)
            .put("float", f)
            .put("double", d)
            .put("bytes", bs)
            .put("age", 30);

    Map<String, FieldAlias> mappings = new HashMap<>();
    mappings.put("long", new FieldAlias("long", true));

    FieldsMappings tm = new FieldsMappings("table", "topic", true, mappings);

    StructFieldsDataExtractor dataExtractor = new StructFieldsDataExtractor(tm);
    StructFieldsDataExtractor.PreparedStatementBinders binders = dataExtractor.get(struct,
            new SinkRecord("", 2, null, null, schema, struct, 2));

    HashMap<String, PreparedStatementBinder> map = new HashMap<>();
    for (PreparedStatementBinder p : Iterables.concat(binders.getNonKeyColumns(), binders.getKeyColumns()))
      map.put(p.getFieldName(), p);

    assertTrue(!binders.isEmpty());
    assertEquals(map.size(), 10);

    assertTrue(Objects.equals(binders.getKeyColumns().get(0).getFieldName(), "long"));

    assertTrue(map.containsKey("firstName"));
    assertTrue(map.get("firstName").getClass() == StringPreparedStatementBinder.class);

    assertTrue(map.containsKey("lastName"));
    assertTrue(map.get("lastName").getClass() == StringPreparedStatementBinder.class);

    assertTrue(map.containsKey("age"));
    assertTrue(map.get("age").getClass() == IntPreparedStatementBinder.class);

    assertTrue(map.get("long").getClass() == LongPreparedStatementBinder.class);
    assertEquals(((LongPreparedStatementBinder) map.get("long")).getValue(), l);

    assertTrue(map.get("short").getClass() == ShortPreparedStatementBinder.class);
    assertEquals(((ShortPreparedStatementBinder) map.get("short")).getValue(), s);

    assertTrue(map.get("byte").getClass() == BytePreparedStatementBinder.class);
    assertEquals(((BytePreparedStatementBinder) map.get("byte")).getValue(), b);

    assertTrue(map.get("float").getClass() == FloatPreparedStatementBinder.class);
    assertEquals(Float.compare(((FloatPreparedStatementBinder) map.get("float")).getValue(), f), 0);

    assertTrue(map.get("double").getClass() == DoublePreparedStatementBinder.class);
    assertEquals(Double.compare(((DoublePreparedStatementBinder) map.get("double")).getValue(), d), 0);

    assertTrue(map.get("bytes").getClass() == BytesPreparedStatementBinder.class);
    assertTrue(Arrays.equals(bs, ((BytesPreparedStatementBinder) map.get("bytes")).getValue()));
  } 
                            #location 51 
                            #vulnerability type NULL_DEREFERENCE