kettle创建mysql连接池

import org.pentaho.di.core.database.Database;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.logging.LogChannel;
import org.pentaho.di.core.logging.LogChannelInterface;
import org.pentaho.di.core.logging.LogLevel;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;

public class MySQLConnectionPoolExample extends BaseStep implements StepInterface {

    private DatabaseMeta dbMeta;
    private Database db;

    public MySQLConnectionPoolExample(StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta, Trans trans) {
        super(stepMeta, stepDataInterface, copyNr, transMeta, trans);
    }

    @Override
    public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
        LogChannel log = new LogChannel(this);
        log.setLogLevel(LogLevel.DETAILED);

        if (first) {
            dbMeta = new DatabaseMeta();
            dbMeta.setName("MySQL Connection Pool");
            dbMeta.setDatabaseType("MySQL");
            dbMeta.setHostname("localhost");
            dbMeta.setPort("3306");
            dbMeta.setDatabaseName("testdb");
            dbMeta.setUsername("user");
            dbMeta.setPassword("password");

            try {
                db = new Database(null, dbMeta);
                db.connect();
            } catch (KettleDatabaseException e) {
                log.logError("Failed to connect to MySQL database", e);
                setErrors(1);
                stopAll();
            }
        }

        // Your data processing logic here

        return true;
    }

    @Override
    public void dispose(StepDataInterface sdi) {
        if (db != null) {
            db.disconnect();
        }
        super.dispose(sdi);
    }
}