class DistributionChannel implements Channel {

    @Override
    public void pushRecord(final Record record) throws InterruptedException {
        if (FinishedRecord.class.equals(record.getClass())) {
            // broadcast
            for (Entry<String, MemoryChannel> entry : channels.entrySet()) {
                entry.getValue().pushRecord(record);
            }
        } else if (DataRecord.class.equals(record.getClass())) {
            toBeAcknowledgeRecords.add(record);
            // hash by table name
            DataRecord dataRecord = (DataRecord) record;
            String index = Integer.toString(Math.abs(dataRecord.hashCode()) % channelNumber);
            channels.get(index).pushRecord(dataRecord);
        } else if (PlaceholderRecord.class.equals(record.getClass())) {
            toBeAcknowledgeRecords.add(record);
            pendingAcknowledgeRecords.put(record.getPosition(), record);
        } else {
            throw new RuntimeException("Not Support Record Type");
        }
    }

    public  DistributionChannel(final int channelNumber, final AckCallback ackCallback);

}