public MessageId initialize() {
        try (Reader<byte[]> reader = WorkerUtils.createReader(
                workerService.getClient().newReader(),
                workerConfig.getWorkerId() + "-function-assignment-initialize",
                workerConfig.getFunctionAssignmentTopic(),
                MessageId.earliest)) {

            // start init phase
            this.isInitializePhase = true;
            // keep track of the last message read
            MessageId lastMessageRead = MessageId.earliest;
            // read all existing messages
            while (reader.hasMessageAvailable()) {
                Message<byte[]> message = reader.readNext();
                lastMessageRead = message.getMessageId();
                processAssignmentMessage(message);
            }
            // init phase is done
            this.isInitializePhase = false;
            // realize existing assignments
            Map<String, Assignment> assignmentMap = workerIdToAssignments.get(this.workerConfig.getWorkerId());
            if (assignmentMap != null) {
                for (Assignment assignment : assignmentMap.values()) {
                    if (needsStart(assignment)) {
                        startFunctionInstance(assignment);
                    }
                }
            }
            // complete future to indicate initialization is complete
            isInitialized.complete(null);
            return lastMessageRead;
        } catch (Exception e) {
            log.error("Failed to initialize function runtime manager: {}", e.getMessage(), e);
            throw new RuntimeException(e);
        }
    }
@Test
    public void testKubernetesFunctionInstancesRestart() throws Exception {

        WorkerConfig workerConfig = new WorkerConfig();
        workerConfig.setWorkerId("worker-1");
        workerConfig.setPulsarServiceUrl(PULSAR_SERVICE_URL);
        workerConfig.setStateStorageServiceUrl("foo");
        workerConfig.setFunctionAssignmentTopicName("assignments");
        WorkerConfig.KubernetesContainerFactory kubernetesContainerFactory
                = new WorkerConfig.KubernetesContainerFactory();
        workerConfig.setKubernetesContainerFactory(kubernetesContainerFactory);
        try (MockedConstruction<KubernetesRuntimeFactory> mocked = Mockito.mockConstruction(KubernetesRuntimeFactory.class,
                (mockedKubernetesRuntimeFactory, context) -> {
                    doNothing().when(mockedKubernetesRuntimeFactory).initialize(
                            any(WorkerConfig.class),
                            any(AuthenticationConfig.class),
                            any(SecretsProviderConfigurator.class),
                            any(),
                            any(),
                            any(),
                            any()
                    );
                    doNothing().when(mockedKubernetesRuntimeFactory).setupClient();
                    doReturn(true).when(mockedKubernetesRuntimeFactory).externallyManaged();

                })) {

            PulsarWorkerService workerService = mock(PulsarWorkerService.class);
            // mock pulsarAdmin sources sinks functions
            PulsarAdmin pulsarAdmin = mock(PulsarAdmin.class);
            Sources sources = mock(Sources.class);
            doNothing().when(sources)
                    .restartSource(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any());
            doReturn(sources).when(pulsarAdmin).sources();
            Sinks sinks = mock(Sinks.class);
            doReturn(sinks).when(pulsarAdmin).sinks();
            Functions functions = mock(Functions.class);
            doNothing().when(functions)
                    .restartFunction(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any());
            doReturn(functions).when(pulsarAdmin).functions();

            doReturn(pulsarAdmin).when(workerService).getFunctionAdmin();
            try (final MockedStatic<RuntimeFactory> runtimeFactoryMockedStatic = Mockito
                    .mockStatic(RuntimeFactory.class);) {

                mockRuntimeFactory(runtimeFactoryMockedStatic);

                List<WorkerInfo> workerInfos = new LinkedList<>();
                workerInfos.add(WorkerInfo.of("worker-1", "localhost", 0));
                workerInfos.add(WorkerInfo.of("worker-2", "localhost", 0));

                MembershipManager membershipManager = mock(MembershipManager.class);
                doReturn(workerInfos).when(membershipManager).getCurrentMembership();

                // build three types of FunctionMetaData
                Function.FunctionMetaData function = Function.FunctionMetaData.newBuilder().setFunctionDetails(
                        Function.FunctionDetails.newBuilder()
                                .setTenant("test-tenant").setNamespace("test-namespace").setName("function")
                                .setComponentType(Function.FunctionDetails.ComponentType.FUNCTION)).build();
                Function.FunctionMetaData source = Function.FunctionMetaData.newBuilder().setFunctionDetails(
                        Function.FunctionDetails.newBuilder()
                                .setTenant("test-tenant").setNamespace("test-namespace").setName("source")
                                .setComponentType(Function.FunctionDetails.ComponentType.SOURCE)).build();
                Function.FunctionMetaData sink = Function.FunctionMetaData.newBuilder().setFunctionDetails(
                        Function.FunctionDetails.newBuilder()
                                .setTenant("test-tenant").setNamespace("test-namespace").setName("sink")
                                .setComponentType(Function.FunctionDetails.ComponentType.SINK)).build();

                @Cleanup
                FunctionRuntimeManager functionRuntimeManager = spy(new FunctionRuntimeManager(
                        workerConfig,
                        workerService,
                        mock(Namespace.class),
                        membershipManager,
                        mock(ConnectorsManager.class),
                        mock(FunctionsManager.class),
                        mock(FunctionMetaDataManager.class),
                        mock(WorkerStatsManager.class),
                        mock(ErrorNotifier.class)));

                // verify restart function/source/sink using different assignment
                verifyRestart(functionRuntimeManager, function, "worker-1", true, false);
                verifyRestart(functionRuntimeManager, function, "worker-2", true, true);
                verifyRestart(functionRuntimeManager, source, "worker-1", true, false);
                verifyRestart(functionRuntimeManager, source, "worker-2", true, true);
                verifyRestart(functionRuntimeManager, sink, "worker-1", true, false);
                verifyRestart(functionRuntimeManager, sink, "worker-2", true, true);
            }
        }
    }