public HdfsBolt withPartitioner(Partitioner partitioner) {
        this.partitioner = partitioner;
        return this;
    }
@Test
    public void testPartitionedOutput() throws IOException {
        HdfsBolt bolt = makeHdfsBolt(hdfsURI, 1, 1000f);

        Partitioner partitoner = new Partitioner() {
            @Override
            public String getPartitionPath(Tuple tuple) {
                return Path.SEPARATOR + tuple.getStringByField("city");
            }
        };

        bolt.prepare(new Config(), topologyContext, collector);
        bolt.withPartitioner(partitoner);

        bolt.execute(tuple1);
        bolt.execute(tuple2);

        verify(collector).ack(tuple1);
        verify(collector).ack(tuple2);

        assertEquals(1, countNonZeroLengthFiles(testRoot + "/SFO"));
        assertEquals(1, countNonZeroLengthFiles(testRoot + "/SJO"));
    }