Separate ListColumnIncluderator cache key parts with nul bytes

processing/src/main/java/io/druid/query/metadata/metadata/ListColumnIncluderator.java

@@ -64,12 +64,13 @@ public class ListColumnIncluderator implements ColumnIncluderator
     for (String column : this.columns) {
       final byte[] bytes = StringUtils.toUtf8(column);
       columns.add(bytes);
-      size += bytes.length;
+      size += bytes.length + 1;
     }
 
     final ByteBuffer bytes = ByteBuffer.allocate(size).put(LIST_CACHE_PREFIX);
     for (byte[] column : columns) {
       bytes.put(column);
+      bytes.put((byte) 0xff);
     }
 
     return bytes.array();

processing/src/test/java/io/druid/query/metadata/SegmentMetadataQueryTest.java

@@ -375,4 +375,26 @@ public class SegmentMetadataQueryTest
       Assert.assertEquals(expectedSegments2.get(i).getInterval(), filteredSegments2.get(i).getInterval());
     }
   }
+
+  @Test
+  public void testCacheKeyWithListColumnIncluderator()
+  {
+    SegmentMetadataQuery oneColumnQuery = Druids.newSegmentMetadataQueryBuilder()
+                                                .dataSource("testing")
+                                                .toInclude(new ListColumnIncluderator(Arrays.asList("foo")))
+                                                .build();
+
+    SegmentMetadataQuery twoColumnQuery = Druids.newSegmentMetadataQueryBuilder()
+                                                .dataSource("testing")
+                                                .toInclude(new ListColumnIncluderator(Arrays.asList("fo", "o")))
+                                                .build();
+
+    final byte[] oneColumnQueryCacheKey = new SegmentMetadataQueryQueryToolChest(null).getCacheStrategy(oneColumnQuery)
+                                                                                      .computeCacheKey(oneColumnQuery);
+
+    final byte[] twoColumnQueryCacheKey = new SegmentMetadataQueryQueryToolChest(null).getCacheStrategy(twoColumnQuery)
+                                                                                      .computeCacheKey(twoColumnQuery);
+
+    Assert.assertFalse(Arrays.equals(oneColumnQueryCacheKey, twoColumnQueryCacheKey));
+  }
 }