private Instant getRecords(Instant lastConsumeInstant, List<Message> messageList, List<T> records)
            throws Exception {
        if (messageList != null) {
            lastConsumeInstant = Instant.now();
            if (!innerFormat) {
                for (Message message : messageList) {
                    T record = deserializationSchema.deserialize(message);
                    records.add(record);
                }
            } else {
                List<RowData> rowDataList = new ArrayList<>();
                ListCollector<RowData> out = new ListCollector<>(rowDataList);
                for (Message message : messageList) {
                    deserializationSchema.deserialize(message, (Collector<T>) out);
                }
                rowDataList.forEach(data -> records.add((T) data));
            }
        }

        return lastConsumeInstant;

    }
Build a TubeMQ source function

@param masterAddress         the master address of TubeMQ
@param topic                 the topic name
@param streamIdSet                the  topic's filter condition items
@param consumerGroup         the consumer group name
@param deserializationSchema the deserialize schema
@param configuration         the configure
@param sessionKey            the tube session key
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/FlinkTubeMQConsumer.java