@Override
    public boolean init(Object srcManager, InstanceProfile srcProfile) {
        try {
            instanceManager = (InstanceManager) srcManager;
            profile = srcProfile;
            auditVersion = Long.parseLong(srcProfile.get(TASK_AUDIT_VERSION));
            setInodeInfo(profile);
            LOGGER.info("task id: {} submit new instance {} profile detail {}.", profile.getTaskId(),
                    profile.getInstanceId(), profile.toJsonStr());
            source = (Source) Class.forName(profile.getSourceClass()).newInstance();
            source.init(profile);
            sink = (Sink) Class.forName(profile.getSinkClass()).newInstance();
            sink.init(profile);
            inited = true;
            return true;
        } catch (Throwable e) {
            handleDeleted();
            doChangeState(State.FATAL);
            LOGGER.error("init instance {} for task {} failed", profile.getInstanceId(), profile.getInstanceId(),
                    e);
            ThreadUtils.threadThrowableHandler(Thread.currentThread(), e);
            return false;
        }
    }
common instance contains source and sink.
main job is to read from source and write to sink
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/instance/CommonInstance.java