- Bytes value = new Bytes(stringSerializer.serialize(TOPIC, Integer.toString(message[i]++)));
- send(key, value, fail(logicErrors, pass, counter), messageSender);
+ send(
+ key,
+ new Bytes(stringSerializer.serialize(TOPIC, "{\"next\":" + message[i]++ + "}")),
+ Message.Type.ADD,
+ poisonPill(poisonPills, pass, counter),
+ logicError(logicErrors, pass, counter),
+ messageSender);