public void startFunction(FunctionRuntimeInfo functionRuntimeInfo) {
        try {
            FunctionMetaData functionMetaData = functionRuntimeInfo.getFunctionInstance().getFunctionMetaData();
            FunctionDetails functionDetails = functionMetaData.getFunctionDetails();
            int instanceId = functionRuntimeInfo.getFunctionInstance().getInstanceId();

            log.info("{}/{}/{}-{} Starting function ...", functionDetails.getTenant(), functionDetails.getNamespace(),
                    functionDetails.getName(), instanceId);

            String packageFile;
            String transformFunctionPackageFile = null;

            Function.PackageLocationMetaData pkgLocation = functionMetaData.getPackageLocation();
            Function.PackageLocationMetaData transformFunctionPkgLocation =
                    functionMetaData.getTransformFunctionPackageLocation();

            if (runtimeFactory.externallyManaged()) {
                packageFile = pkgLocation.getPackagePath();
                transformFunctionPackageFile = transformFunctionPkgLocation.getPackagePath();
            } else {
                packageFile = getPackageFile(functionMetaData, functionDetails, instanceId, pkgLocation,
                        InstanceUtils.calculateSubjectType(functionDetails));
                if (!isEmpty(transformFunctionPkgLocation.getPackagePath())) {
                    transformFunctionPackageFile =
                            getPackageFile(functionMetaData, functionDetails, instanceId, transformFunctionPkgLocation,
                                    FunctionDetails.ComponentType.FUNCTION);
                }
            }

            // Setup for batch sources if necessary
            setupBatchSource(functionDetails);

            RuntimeSpawner runtimeSpawner = getRuntimeSpawner(functionRuntimeInfo.getFunctionInstance(),
                    packageFile, transformFunctionPackageFile);
            functionRuntimeInfo.setRuntimeSpawner(runtimeSpawner);

            runtimeSpawner.start();
        } catch (Exception ex) {
            FunctionDetails details = functionRuntimeInfo.getFunctionInstance()
                    .getFunctionMetaData().getFunctionDetails();
            log.error("{}/{}/{} Error starting function", details.getTenant(), details.getNamespace(),
                    details.getName(), ex);
            functionRuntimeInfo.setStartupException(ex);
        }
    }
@Test
    public void testStartFunctionWithDLNamespace() throws Exception {

        WorkerConfig workerConfig = new WorkerConfig();
        workerConfig.setWorkerId("worker-1");
        workerConfig.setFunctionRuntimeFactoryClassName(ThreadRuntimeFactory.class.getName());
        workerConfig.setFunctionRuntimeFactoryConfigs(
                ObjectMapperFactory.getMapper().getObjectMapper().convertValue(
                        new ThreadRuntimeFactoryConfig().setThreadGroupName("test"), Map.class));
        workerConfig.setPulsarServiceUrl("pulsar://localhost:6650");
        workerConfig.setStateStorageServiceUrl("foo");
        workerConfig.setFunctionAssignmentTopicName("assignments");

        ThreadRuntimeFactory factory = mock(ThreadRuntimeFactory.class);
        Namespace dlogNamespace = mock(Namespace.class);
        // throw exception when dlogNamespace is accessed by actioner and verify it
        final String exceptionMsg = "dl namespace not-found";
        doThrow(new IllegalArgumentException(exceptionMsg)).when(dlogNamespace).openLog(any());

        @SuppressWarnings("resource")
        FunctionActioner actioner = new FunctionActioner(workerConfig, factory, dlogNamespace,
                new ConnectorsManager(workerConfig), new FunctionsManager(workerConfig), mock(PulsarAdmin.class),
                mock(PackageUrlValidator.class));
        Function.FunctionMetaData function1 = Function.FunctionMetaData.newBuilder()
                .setFunctionDetails(Function.FunctionDetails.newBuilder().setTenant("test-tenant")
                        .setNamespace("test-namespace").setName("func-1"))
                .build();
        Function.Instance instance = Function.Instance.newBuilder().setFunctionMetaData(function1).setInstanceId(0)
                .build();
        FunctionRuntimeInfo functionRuntimeInfo = mock(FunctionRuntimeInfo.class);
        doReturn(instance).when(functionRuntimeInfo).getFunctionInstance();
        doThrow(new IllegalStateException("StartupException")).when(functionRuntimeInfo).setStartupException(any());

        // actioner should try to download file from bk-dlogNamespace and fails with exception
        try {
            actioner.startFunction(functionRuntimeInfo);
            fail();
        } catch (IllegalStateException ex) {
            assertEquals(ex.getMessage(), "StartupException");
        }
    }