@Override
    public int channel(CommittableMessage<MultiTableCommittable> committableMessage) {
        if (committableMessage instanceof CommittableWithLineage) {
            MultiTableCommittable multiTableCommittable =
                    ((CommittableWithLineage<MultiTableCommittable>) committableMessage)
                            .getCommittable();
            return Math.floorMod(
                    Objects.hash(
                            multiTableCommittable.getDatabase(), multiTableCommittable.getTable()),
                    numChannels);
        } else {
            // CommittableSummary is inaccurate after shuffling, need to be recreated.
            return Math.floorMod(Objects.hash(committableMessage), numChannels);
        }
    }
@Test
    public void testChannel() {
        MultiTableCommittableChannelComputer computer = new MultiTableCommittableChannelComputer();
        computer.setup(4);
        List<MultiTableCommittable> commits =
                Arrays.asList(
                        new MultiTableCommittable("database", "table1", 1L, null, null),
                        new MultiTableCommittable("database", "table2", 1L, null, null),
                        new MultiTableCommittable("database", "table1", 1L, null, null),
                        new MultiTableCommittable("database", "table5", 1L, null, null),
                        new MultiTableCommittable("database", "table3", 1L, null, null),
                        new MultiTableCommittable("database", "table8", 1L, null, null),
                        new MultiTableCommittable("database", "table5", 1L, null, null),
                        new MultiTableCommittable("database", "table1", 1L, null, null),
                        new MultiTableCommittable("database", "table9", 1L, null, null),
                        new MultiTableCommittable("database", "table5", 1L, null, null),
                        new MultiTableCommittable("database", "table3", 1L, null, null),
                        new MultiTableCommittable("database", "table8", 1L, null, null));
        Map<Integer, Set<String>> map = new HashMap<>();
        commits.forEach(
                (commit) -> {
                    int channel = computer.channel(new CommittableWithLineage<>(commit, 1L, 0));
                    Set<String> set = map.getOrDefault(channel, new HashSet<>());
                    set.add(commit.getTable());
                    map.put(channel, set);
                });
        Set<String> actualtables = new HashSet<>();
        for (Map.Entry<Integer, Set<String>> entry : map.entrySet()) {
            actualtables.addAll(entry.getValue());
        }
        Set<String> expectedTables =
                new HashSet<>(
                        Arrays.asList("table1", "table2", "table3", "table5", "table8", "table9"));
        // Not a table is appeared in more than one channel.
        Assertions.assertEquals(actualtables, expectedTables);
    }