queue.process(builder.build());
for (int i = 0; i < 15; ++i) {
builder.withMessage("topic3", Integer.toString(i).getBytes());
}
queue.process(builder.build());
// sink3: 15 with topic3_alias
for (int i = 0; i < 20; ++i) {
builder.withMessage("topic4", Integer.toString(i).getBytes());
}