@Test public void testStartFunctionWithDLNamespace() throws Exception { WorkerConfig workerConfig = new WorkerConfig(); workerConfig.setWorkerId("worker-1"); workerConfig.setFunctionRuntimeFactoryClassName(ThreadRuntimeFactory.class.getName()); workerConfig.setFunctionRuntimeFactoryConfigs( ObjectMapperFactory.getThreadLocal().convertValue( new ThreadRuntimeFactoryConfig().setThreadGroupName("test"), Map.class)); workerConfig.setPulsarServiceUrl("pulsar: workerConfig.setStateStorageServiceUrl("foo"); workerConfig.setFunctionAssignmentTopicName("assignments"); ThreadRuntimeFactory factory = mock(ThreadRuntimeFactory.class); Namespace dlogNamespace = mock(Namespace.class); 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)); Runtime runtime = mock(Runtime.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()); try { actioner.startFunction(functionRuntimeInfo); fail(); } catch (IllegalStateException ex) { assertEquals(ex.getMessage(), "StartupException"); } }
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 pkgLocation = functionMetaData.getPackageLocation().getPackagePath(); boolean isPkgUrlProvided = isFunctionPackageUrlSupported(pkgLocation); if (runtimeFactory.externallyManaged()) { packageFile = pkgLocation; } else { if (isPkgUrlProvided && pkgLocation.startsWith(FILE)) { URL url = new URL(pkgLocation); File pkgFile = new File(url.toURI()); packageFile = pkgFile.getAbsolutePath(); } else if (WorkerUtils.isFunctionCodeBuiltin(functionDetails)) { File pkgFile = getBuiltinArchive(FunctionDetails.newBuilder(functionMetaData.getFunctionDetails())); packageFile = pkgFile.getAbsolutePath(); } else { File pkgDir = new File(workerConfig.getDownloadDirectory(), getDownloadPackagePath(functionMetaData, instanceId)); pkgDir.mkdirs(); File pkgFile = new File( pkgDir, new File(getDownloadFileName(functionMetaData.getFunctionDetails(), functionMetaData.getPackageLocation())).getName()); downloadFile(pkgFile, isPkgUrlProvided, functionMetaData, instanceId); packageFile = pkgFile.getAbsolutePath(); } } setupBatchSource(functionDetails); RuntimeSpawner runtimeSpawner = getRuntimeSpawner(functionRuntimeInfo.getFunctionInstance(), packageFile); functionRuntimeInfo.setRuntimeSpawner(runtimeSpawner); runtimeSpawner.start(); } catch (Exception ex) { FunctionDetails details = functionRuntimeInfo.getFunctionInstance() .getFunctionMetaData().getFunctionDetails(); log.info("{}/{}/{} Error starting function", details.getTenant(), details.getNamespace(), details.getName(), ex); functionRuntimeInfo.setStartupException(ex); } }
FunctionActioner { 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 pkgLocation = functionMetaData.getPackageLocation().getPackagePath(); boolean isPkgUrlProvided = isFunctionPackageUrlSupported(pkgLocation); if (runtimeFactory.externallyManaged()) { packageFile = pkgLocation; } else { if (isPkgUrlProvided && pkgLocation.startsWith(FILE)) { URL url = new URL(pkgLocation); File pkgFile = new File(url.toURI()); packageFile = pkgFile.getAbsolutePath(); } else if (WorkerUtils.isFunctionCodeBuiltin(functionDetails)) { File pkgFile = getBuiltinArchive(FunctionDetails.newBuilder(functionMetaData.getFunctionDetails())); packageFile = pkgFile.getAbsolutePath(); } else { File pkgDir = new File(workerConfig.getDownloadDirectory(), getDownloadPackagePath(functionMetaData, instanceId)); pkgDir.mkdirs(); File pkgFile = new File( pkgDir, new File(getDownloadFileName(functionMetaData.getFunctionDetails(), functionMetaData.getPackageLocation())).getName()); downloadFile(pkgFile, isPkgUrlProvided, functionMetaData, instanceId); packageFile = pkgFile.getAbsolutePath(); } } setupBatchSource(functionDetails); RuntimeSpawner runtimeSpawner = getRuntimeSpawner(functionRuntimeInfo.getFunctionInstance(), packageFile); functionRuntimeInfo.setRuntimeSpawner(runtimeSpawner); runtimeSpawner.start(); } catch (Exception ex) { FunctionDetails details = functionRuntimeInfo.getFunctionInstance() .getFunctionMetaData().getFunctionDetails(); log.info("{}/{}/{} Error starting function", details.getTenant(), details.getNamespace(), details.getName(), ex); functionRuntimeInfo.setStartupException(ex); } } }
FunctionActioner { 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 pkgLocation = functionMetaData.getPackageLocation().getPackagePath(); boolean isPkgUrlProvided = isFunctionPackageUrlSupported(pkgLocation); if (runtimeFactory.externallyManaged()) { packageFile = pkgLocation; } else { if (isPkgUrlProvided && pkgLocation.startsWith(FILE)) { URL url = new URL(pkgLocation); File pkgFile = new File(url.toURI()); packageFile = pkgFile.getAbsolutePath(); } else if (WorkerUtils.isFunctionCodeBuiltin(functionDetails)) { File pkgFile = getBuiltinArchive(FunctionDetails.newBuilder(functionMetaData.getFunctionDetails())); packageFile = pkgFile.getAbsolutePath(); } else { File pkgDir = new File(workerConfig.getDownloadDirectory(), getDownloadPackagePath(functionMetaData, instanceId)); pkgDir.mkdirs(); File pkgFile = new File( pkgDir, new File(getDownloadFileName(functionMetaData.getFunctionDetails(), functionMetaData.getPackageLocation())).getName()); downloadFile(pkgFile, isPkgUrlProvided, functionMetaData, instanceId); packageFile = pkgFile.getAbsolutePath(); } } setupBatchSource(functionDetails); RuntimeSpawner runtimeSpawner = getRuntimeSpawner(functionRuntimeInfo.getFunctionInstance(), packageFile); functionRuntimeInfo.setRuntimeSpawner(runtimeSpawner); runtimeSpawner.start(); } catch (Exception ex) { FunctionDetails details = functionRuntimeInfo.getFunctionInstance() .getFunctionMetaData().getFunctionDetails(); log.info("{}/{}/{} Error starting function", details.getTenant(), details.getNamespace(), details.getName(), ex); functionRuntimeInfo.setStartupException(ex); } } FunctionActioner(WorkerConfig workerConfig,
                            RuntimeFactory runtimeFactory,
                            Namespace dlogNamespace,
                            ConnectorsManager connectorsManager,FunctionsManager functionsManager,PulsarAdmin pulsarAdmin); }
FunctionActioner { 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 pkgLocation = functionMetaData.getPackageLocation().getPackagePath(); boolean isPkgUrlProvided = isFunctionPackageUrlSupported(pkgLocation); if (runtimeFactory.externallyManaged()) { packageFile = pkgLocation; } else { if (isPkgUrlProvided && pkgLocation.startsWith(FILE)) { URL url = new URL(pkgLocation); File pkgFile = new File(url.toURI()); packageFile = pkgFile.getAbsolutePath(); } else if (WorkerUtils.isFunctionCodeBuiltin(functionDetails)) { File pkgFile = getBuiltinArchive(FunctionDetails.newBuilder(functionMetaData.getFunctionDetails())); packageFile = pkgFile.getAbsolutePath(); } else { File pkgDir = new File(workerConfig.getDownloadDirectory(), getDownloadPackagePath(functionMetaData, instanceId)); pkgDir.mkdirs(); File pkgFile = new File( pkgDir, new File(getDownloadFileName(functionMetaData.getFunctionDetails(), functionMetaData.getPackageLocation())).getName()); downloadFile(pkgFile, isPkgUrlProvided, functionMetaData, instanceId); packageFile = pkgFile.getAbsolutePath(); } } setupBatchSource(functionDetails); RuntimeSpawner runtimeSpawner = getRuntimeSpawner(functionRuntimeInfo.getFunctionInstance(), packageFile); functionRuntimeInfo.setRuntimeSpawner(runtimeSpawner); runtimeSpawner.start(); } catch (Exception ex) { FunctionDetails details = functionRuntimeInfo.getFunctionInstance() .getFunctionMetaData().getFunctionDetails(); log.info("{}/{}/{} Error starting function", details.getTenant(), details.getNamespace(), details.getName(), ex); functionRuntimeInfo.setStartupException(ex); } } FunctionActioner(WorkerConfig workerConfig,
                            RuntimeFactory runtimeFactory,
                            Namespace dlogNamespace,
                            ConnectorsManager connectorsManager,FunctionsManager functionsManager,PulsarAdmin pulsarAdmin); void startFunction(FunctionRuntimeInfo functionRuntimeInfo); void stopFunction(FunctionRuntimeInfo functionRuntimeInfo); void terminateFunction(FunctionRuntimeInfo functionRuntimeInfo); }
FunctionActioner { 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 pkgLocation = functionMetaData.getPackageLocation().getPackagePath(); boolean isPkgUrlProvided = isFunctionPackageUrlSupported(pkgLocation); if (runtimeFactory.externallyManaged()) { packageFile = pkgLocation; } else { if (isPkgUrlProvided && pkgLocation.startsWith(FILE)) { URL url = new URL(pkgLocation); File pkgFile = new File(url.toURI()); packageFile = pkgFile.getAbsolutePath(); } else if (WorkerUtils.isFunctionCodeBuiltin(functionDetails)) { File pkgFile = getBuiltinArchive(FunctionDetails.newBuilder(functionMetaData.getFunctionDetails())); packageFile = pkgFile.getAbsolutePath(); } else { File pkgDir = new File(workerConfig.getDownloadDirectory(), getDownloadPackagePath(functionMetaData, instanceId)); pkgDir.mkdirs(); File pkgFile = new File( pkgDir, new File(getDownloadFileName(functionMetaData.getFunctionDetails(), functionMetaData.getPackageLocation())).getName()); downloadFile(pkgFile, isPkgUrlProvided, functionMetaData, instanceId); packageFile = pkgFile.getAbsolutePath(); } } setupBatchSource(functionDetails); RuntimeSpawner runtimeSpawner = getRuntimeSpawner(functionRuntimeInfo.getFunctionInstance(), packageFile); functionRuntimeInfo.setRuntimeSpawner(runtimeSpawner); runtimeSpawner.start(); } catch (Exception ex) { FunctionDetails details = functionRuntimeInfo.getFunctionInstance() .getFunctionMetaData().getFunctionDetails(); log.info("{}/{}/{} Error starting function", details.getTenant(), details.getNamespace(), details.getName(), ex); functionRuntimeInfo.setStartupException(ex); } } FunctionActioner(WorkerConfig workerConfig,
                            RuntimeFactory runtimeFactory,
                            Namespace dlogNamespace,
                            ConnectorsManager connectorsManager,FunctionsManager functionsManager,PulsarAdmin pulsarAdmin); void startFunction(FunctionRuntimeInfo functionRuntimeInfo); void stopFunction(FunctionRuntimeInfo functionRuntimeInfo); void terminateFunction(FunctionRuntimeInfo functionRuntimeInfo);  }