private void addToMemory(InstanceProfile instanceProfile) {
        String inlongGroupId = instanceProfile.getInlongGroupId();
        String inlongStreamId = instanceProfile.getInlongStreamId();
        Instance oldInstance = instanceMap.get(instanceProfile.getInstanceId());
        if (oldInstance != null) {
            oldInstance.destroy();
            instanceMap.remove(instanceProfile.getInstanceId());
            LOGGER.error("old instance {} should not exist, try stop it first",
                    instanceProfile.getInstanceId());
            AuditUtils.add(AuditUtils.AUDIT_ID_AGENT_DEL_INSTANCE_MEM_UNUSUAL, inlongGroupId, inlongStreamId,
                    instanceProfile.getSinkDataTime(), 1, 1, auditVersion);
        }
        LOGGER.info("instanceProfile {}", instanceProfile.toJsonStr());
        try {
            if (taskManager != null && taskManager.getInstanceNum() > globalInstanceLimit) {
                LOGGER.error("global instance num {} over limit {}", taskManager.getInstanceNum(), globalInstanceLimit);
                return;
            }
            if (instanceMap.size() > instanceLimit) {
                LOGGER.info("add instance to memory refused because instanceMap size over limit {}",
                        instanceProfile.getInstanceId());
                return;
            }
            Class<?> taskClass = Class.forName(instanceProfile.getInstanceClass());
            Instance instance = (Instance) taskClass.newInstance();
            boolean initSuc = instance.init(this, instanceProfile);
            if (initSuc) {
                instanceMap.put(instanceProfile.getInstanceId(), instance);
                EXECUTOR_SERVICE.submit(instance);
                LOGGER.info(
                        "add instance to memory instanceId {} instanceMap size {}, runningPool instance total {}, runningPool instance active {}",
                        instance.getInstanceId(), instanceMap.size(), EXECUTOR_SERVICE.getTaskCount(),
                        EXECUTOR_SERVICE.getActiveCount());
                AuditUtils.add(AuditUtils.AUDIT_ID_AGENT_ADD_INSTANCE_MEM, inlongGroupId, inlongStreamId,
                        instanceProfile.getSinkDataTime(), 1, 1, auditVersion);
            } else {
                LOGGER.error(
                        "add instance to memory init failed instanceId {}", instance.getInstanceId());
                AuditUtils.add(AuditUtils.AUDIT_ID_AGENT_ADD_INSTANCE_MEM_FAILED, inlongGroupId, inlongStreamId,
                        instanceProfile.getSinkDataTime(), 1, 1, auditVersion);
            }
        } catch (Throwable t) {
            LOGGER.error("add instance error.", t);
        }
    }
add instance to memory, if there is a record refer to the instance id exist we need to destroy it first.
inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/instance/InstanceManager.java