Iterator<SourceRecords> splitSchemaChangeStream(List<DataChangeEvent> batchEvents) {
        return new SchemaChangeStreamSplitter().split(batchEvents);
    }
@Test
    public void testSplitSchemaChangeStream() throws Exception {
        IncrementalSourceStreamFetcher fetcher = createFetcher();

        List<DataChangeEvent> inputEvents = new ArrayList<>();
        List<SourceRecords> records = new ArrayList<>();
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        Iterator<SourceRecords> outputEvents = fetcher.splitSchemaChangeStream(inputEvents);
        outputEvents.forEachRemaining(records::add);

        Assertions.assertEquals(1, records.size());
        Assertions.assertEquals(2, records.get(0).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(0).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(0).getSourceRecordList().get(1)));

        inputEvents = new ArrayList<>();
        records = new ArrayList<>();
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        outputEvents = fetcher.splitSchemaChangeStream(inputEvents);
        outputEvents.forEachRemaining(records::add);

        Assertions.assertEquals(2, records.size());
        Assertions.assertEquals(1, records.get(0).getSourceRecordList().size());
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeBeforeWatermarkEvent(
                        records.get(0).getSourceRecordList().get(0)));
        Assertions.assertEquals(3, records.get(1).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(1).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(1).getSourceRecordList().get(1)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeAfterWatermarkEvent(
                        records.get(1).getSourceRecordList().get(2)));

        inputEvents = new ArrayList<>();
        records = new ArrayList<>();
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeUnknownEvent()));
        outputEvents = fetcher.splitSchemaChangeStream(inputEvents);
        outputEvents.forEachRemaining(records::add);

        Assertions.assertEquals(2, records.size());
        Assertions.assertEquals(3, records.get(0).getSourceRecordList().size());
        Assertions.assertEquals(3, records.get(1).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(0).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(0).getSourceRecordList().get(1)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeBeforeWatermarkEvent(
                        records.get(0).getSourceRecordList().get(2)));
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(1).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(1).getSourceRecordList().get(1)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeAfterWatermarkEvent(
                        records.get(1).getSourceRecordList().get(2)));

        inputEvents = new ArrayList<>();
        records = new ArrayList<>();
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeUnknownEvent()));
        outputEvents = fetcher.splitSchemaChangeStream(inputEvents);
        outputEvents.forEachRemaining(records::add);

        Assertions.assertEquals(3, records.size());
        Assertions.assertEquals(1, records.get(0).getSourceRecordList().size());
        Assertions.assertEquals(3, records.get(1).getSourceRecordList().size());
        Assertions.assertEquals(2, records.get(2).getSourceRecordList().size());
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeBeforeWatermarkEvent(
                        records.get(0).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(1).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(1).getSourceRecordList().get(1)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeAfterWatermarkEvent(
                        records.get(1).getSourceRecordList().get(2)));
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(2).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(2).getSourceRecordList().get(1)));

        inputEvents = new ArrayList<>();
        records = new ArrayList<>();
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        outputEvents = fetcher.splitSchemaChangeStream(inputEvents);
        outputEvents.forEachRemaining(records::add);

        Assertions.assertEquals(3, records.size());
        Assertions.assertEquals(2, records.get(0).getSourceRecordList().size());
        Assertions.assertEquals(3, records.get(1).getSourceRecordList().size());
        Assertions.assertEquals(1, records.get(2).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(0).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeBeforeWatermarkEvent(
                        records.get(0).getSourceRecordList().get(1)));
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(1).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(1).getSourceRecordList().get(1)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeAfterWatermarkEvent(
                        records.get(1).getSourceRecordList().get(2)));
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(2).getSourceRecordList().get(0)));

        inputEvents = new ArrayList<>();
        records = new ArrayList<>();
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        outputEvents = fetcher.splitSchemaChangeStream(inputEvents);
        outputEvents.forEachRemaining(records::add);

        Assertions.assertEquals(4, records.size());
        Assertions.assertEquals(2, records.get(0).getSourceRecordList().size());
        Assertions.assertEquals(2, records.get(1).getSourceRecordList().size());
        Assertions.assertEquals(2, records.get(2).getSourceRecordList().size());
        Assertions.assertEquals(2, records.get(3).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(0).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeBeforeWatermarkEvent(
                        records.get(0).getSourceRecordList().get(1)));
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(1).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeAfterWatermarkEvent(
                        records.get(1).getSourceRecordList().get(1)));
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(2).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeBeforeWatermarkEvent(
                        records.get(2).getSourceRecordList().get(1)));
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(3).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeAfterWatermarkEvent(
                        records.get(3).getSourceRecordList().get(1)));

        inputEvents = new ArrayList<>();
        records = new ArrayList<>();
        inputEvents.add(new DataChangeEvent(createHeartbeatEvent()));
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        inputEvents.add(new DataChangeEvent(createHeartbeatEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        inputEvents.add(new DataChangeEvent(createHeartbeatEvent()));
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        inputEvents.add(new DataChangeEvent(createHeartbeatEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        inputEvents.add(new DataChangeEvent(createHeartbeatEvent()));
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        inputEvents.add(new DataChangeEvent(createSchemaChangeEvent()));
        inputEvents.add(new DataChangeEvent(createDataEvent()));
        inputEvents.add(new DataChangeEvent(createHeartbeatEvent()));
        outputEvents = fetcher.splitSchemaChangeStream(inputEvents);
        outputEvents.forEachRemaining(records::add);

        Assertions.assertEquals(11, records.size());
        Assertions.assertEquals(3, records.get(0).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isHeartbeatRecord(records.get(0).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(0).getSourceRecordList().get(1)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeBeforeWatermarkEvent(
                        records.get(0).getSourceRecordList().get(2)));
        Assertions.assertEquals(2, records.get(1).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(1).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeAfterWatermarkEvent(
                        records.get(1).getSourceRecordList().get(1)));
        Assertions.assertEquals(2, records.get(2).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isHeartbeatRecord(records.get(2).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeBeforeWatermarkEvent(
                        records.get(2).getSourceRecordList().get(1)));
        Assertions.assertEquals(2, records.get(3).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(3).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeAfterWatermarkEvent(
                        records.get(3).getSourceRecordList().get(1)));
        Assertions.assertEquals(3, records.get(4).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(4).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(4).getSourceRecordList().get(1)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeBeforeWatermarkEvent(
                        records.get(4).getSourceRecordList().get(2)));
        Assertions.assertEquals(2, records.get(5).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(5).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeAfterWatermarkEvent(
                        records.get(5).getSourceRecordList().get(1)));
        Assertions.assertEquals(4, records.get(6).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isHeartbeatRecord(records.get(6).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(6).getSourceRecordList().get(1)));
        Assertions.assertTrue(
                SourceRecordUtils.isHeartbeatRecord(records.get(6).getSourceRecordList().get(2)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeBeforeWatermarkEvent(
                        records.get(6).getSourceRecordList().get(3)));
        Assertions.assertEquals(3, records.get(7).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(7).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(7).getSourceRecordList().get(1)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeAfterWatermarkEvent(
                        records.get(7).getSourceRecordList().get(2)));
        Assertions.assertEquals(3, records.get(8).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isHeartbeatRecord(records.get(8).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(8).getSourceRecordList().get(1)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeBeforeWatermarkEvent(
                        records.get(8).getSourceRecordList().get(2)));
        Assertions.assertEquals(2, records.get(9).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isSchemaChangeEvent(records.get(9).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                WatermarkEvent.isSchemaChangeAfterWatermarkEvent(
                        records.get(9).getSourceRecordList().get(1)));
        Assertions.assertEquals(2, records.get(10).getSourceRecordList().size());
        Assertions.assertTrue(
                SourceRecordUtils.isDataChangeRecord(records.get(10).getSourceRecordList().get(0)));
        Assertions.assertTrue(
                SourceRecordUtils.isHeartbeatRecord(records.get(10).getSourceRecordList().get(1)));
    }