public void auditSourceToMysql() {
        DataQueue dataQueue = new DataQueue(queueSize);
        List<JdbcConfig> sourceList = ConfigService.getInstance().getAuditSourceByServiceId(serviceId);
        for (JdbcConfig jdbcConfig : sourceList) {
            JdbcSource jdbcSource = new JdbcSource(dataQueue, buildAuditJdbcSourceConfig(jdbcConfig));
            jdbcSource.start();
            originalSources.add(jdbcSource);
            LOGGER.info("Audit source to mysql jdbc config:{}", jdbcConfig);
        }

        SinkConfig sinkConfig = buildMysqlSinkConfig(configuration.get(KEY_MYSQL_SINK_INSERT_TEMP_SQL,
                DEFAULT_MYSQL_SINK_INSERT_TEMP_SQL));
        JdbcSink sink = new JdbcSink(dataQueue, sinkConfig);
        sink.start();
        dataFlowSinks.add(sink);
    }
Etl service aggregate the data from the data source and store the aggregated data to the target storage.
inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/service/node/EtlService.java