public InternalRowCollector getInternalRowCollector(
            Handover<InternalRow> handover,
            Object checkpointLock,
            Map<String, String> envOptionsInfo) {
        if (isMultiTable) {
            return new InternalMultiRowCollector(
                    handover, checkpointLock, rowSerializationMap, envOptionsInfo);
        } else {
            return new InternalRowCollector(
                    handover, checkpointLock, rowSerialization, envOptionsInfo);
        }
    }
@Test
    public void testMultiReaderConverter() throws IOException {
        initSchema();
        initData();
        MultiTableManager multiTableManager =
                new MultiTableManager(
                        new CatalogTable[] {catalogTable1, catalogTable2, catalogTable3});
        InternalMultiRowCollector internalMultiRowCollector =
                (InternalMultiRowCollector)
                        multiTableManager.getInternalRowCollector(null, null, null);
        Map<String, InternalRowConverter> rowSerializationMap =
                internalMultiRowCollector.getRowSerializationMap();
        InternalRow internalRow =
                rowSerializationMap.get(seaTunnelRow1.getTableId()).convert(seaTunnelRow1);
        for (int v = 0; v < specificInternalRow2.numFields(); v++) {
            if (specificInternalRow2.genericGet(v) instanceof ArrayBasedMapData) {
                Assertions.assertEquals(
                        specificInternalRow2.getMap(v).keyArray(),
                        internalRow.getMap(v).keyArray());
                Assertions.assertEquals(
                        specificInternalRow2.getMap(v).valueArray(),
                        internalRow.getMap(v).valueArray());
            } else if (specificInternalRow2.genericGet(v) instanceof SpecificInternalRow) {
                SpecificInternalRow expected =
                        (SpecificInternalRow) specificInternalRow2.genericGet(v);
                SpecificInternalRow actual =
                        (SpecificInternalRow) ((SpecificInternalRow) internalRow).genericGet(v);
                for (int o = 0; v < expected.numFields(); v++) {
                    if (expected.genericGet(o) instanceof ArrayBasedMapData) {
                        Assertions.assertEquals(
                                expected.getMap(o).keyArray(), actual.getMap(o).keyArray());
                        Assertions.assertEquals(
                                expected.getMap(o).valueArray(), actual.getMap(o).valueArray());
                    } else {
                        Assertions.assertEquals(
                                expected.genericGet(v),
                                ((SpecificInternalRow) actual).genericGet(v));
                    }
                }
            } else {
                Assertions.assertEquals(
                        specificInternalRow2.genericGet(v),
                        ((SpecificInternalRow) internalRow).genericGet(v));
            }
        }
        InternalRow internalRow3 =
                rowSerializationMap.get(seaTunnelRow3.getTableId()).convert(seaTunnelRow3);
        Assertions.assertEquals(specificInternalRow3, internalRow3);
        for (int v = 0; v < specificInternalRow3.numFields(); v++) {
            Assertions.assertEquals(
                    specificInternalRow3.genericGet(v),
                    ((SpecificInternalRow) internalRow3).genericGet(v));
        }
    }