testBulkNPE.java

    @Override
    protected Settings nodeSettings(int nodeOrdinal) {
        return Settings.builder()
            .put(super.nodeSettings(nodeOrdinal))
            .put("threadpool.bulk.type", "fixed")
            .put("threadpool.bulk.queue_size", "500")
            .put("threadpool.bulk.processors", "8")
            .build();
    }

     @Test
    public void testBulkInsert2() throws Exception {
        execute("create table t (" +
                " id integer primary key, " +
                " col1 string," +
                " col2 string," +
                " col3 string," +
                " col4 string" +
                ")");
        ensureYellow();
        int bulkSize = 1000;
        Object[][] bulkArgs = new Object[bulkSize][];
        for (int i = 0; i < bulkSize; i++) {
            bulkArgs[i] = new Object[]{i, ""+i, ""+i, ""+i, ""+i};
        }
        SQLBulkResponse bulkResponse = execute("insert into t (id, col1, col2, col3, col4) values (?, ?, ?, ?, ?)", bulkArgs);
        assertThat(bulkResponse.results().length, is(bulkSize));
        execute("refresh table t");
        // assert that bulk insert has inserted everything it said it has
        execute("select count(*) from t");
        assertEquals((long)bulkSize, response.rows()[0][0]);

        bulkArgs = new Object[bulkSize][];
        for (int i = 0; i < bulkSize; i++) {
            bulkArgs[i] = new Object[]{""+i+1000, ""+i+1000, ""+i+1000, ""+i+1000, i};
        }
        bulkResponse = execute("UPDATE t SET col1 = ?, col2 = ?, col3 = ?, col4 = ? WHERE id = ?", bulkArgs);
        assertThat(bulkResponse.results().length, is(bulkSize));
        execute("refresh table t");
        execute("select count(*) from t");
        assertEquals((long)bulkSize, response.rows()[0][0]);
        execute("select * from t order by 1");
        System.out.println(bulkArgs);
        assertEquals("", TestingHelpers.printedTable(response.rows()));
    }