srkukarni closed pull request #2784: add initialize routine to FunctionRuntimeManager URL: https://github.com/apache/pulsar/pull/2784
This is a PR merged from a forked repository. As GitHub hides the original diff on merge, it is displayed below for the sake of provenance: As this is a foreign pull request (from a fork), the diff is supplied below (as it won't show otherwise due to GitHub magic): diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailer.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailer.java index 3ad6c7c6ee..3f8bec3310 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailer.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailer.java @@ -36,13 +36,11 @@ private final FunctionRuntimeManager functionRuntimeManager; private final Reader<byte[]> reader; - public FunctionAssignmentTailer(FunctionRuntimeManager functionRuntimeManager) + public FunctionAssignmentTailer(FunctionRuntimeManager functionRuntimeManager, Reader<byte[]> reader) throws PulsarClientException { this.functionRuntimeManager = functionRuntimeManager; - this.reader = functionRuntimeManager.getWorkerService().getClient().newReader() - .topic(functionRuntimeManager.getWorkerConfig().getFunctionAssignmentTopic()).readCompacted(true) - .startMessageId(MessageId.earliest).create(); + this.reader = reader; } public void start() { @@ -66,8 +64,7 @@ public void close() { log.info("Stopped function state consumer"); } - @Override - public void accept(Message<byte[]> msg) { + public void processAssignment(Message<byte[]> msg) { if(msg.getData()==null || (msg.getData().length==0)) { log.info("Received assignment delete: {}", msg.getKey()); this.functionRuntimeManager.deleteAssignment(msg.getKey()); @@ -82,8 +79,13 @@ public void accept(Message<byte[]> msg) { throw new RuntimeException(e); } log.info("Received assignment update: {}", assignment); - this.functionRuntimeManager.processAssignment(assignment); + this.functionRuntimeManager.processAssignment(assignment); } + } + + @Override + public void accept(Message<byte[]> msg) { + processAssignment(msg); // receive next request receiveOne(); } diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionMetaDataManager.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionMetaDataManager.java index 44ff807125..4faed11a38 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionMetaDataManager.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionMetaDataManager.java @@ -82,13 +82,11 @@ public FunctionMetaDataManager(WorkerConfig workerConfig, /** * Initializes the FunctionMetaDataManager. Does the following: - * 1. Restores from snapshot if one exists - * 2. Sends out initialize marker to FMT and consume messages until the initialize marker is consumed + * 1. Consume all existing function meta data upon start to establish existing state */ public void initialize() { log.info("/** Initializing Function Metadata Manager **/"); try { - Reader<byte[]> reader = pulsarClient.newReader() .topic(this.workerConfig.getFunctionMetadataTopic()) .startMessageId(MessageId.earliest) diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java index 47d317fc94..8c0b50b849 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java @@ -36,10 +36,13 @@ import javax.ws.rs.core.Response.Status; import javax.ws.rs.core.UriBuilder; +import lombok.Setter; import org.apache.commons.lang3.StringUtils; import org.apache.distributedlog.api.namespace.Namespace; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.Reader; import org.apache.pulsar.common.policies.data.ErrorData; import org.apache.pulsar.functions.instance.AuthenticationConfig; import org.apache.pulsar.functions.proto.Function.Assignment; @@ -65,6 +68,8 @@ // All the runtime info related to functions executed by this worker // Fully Qualified InstanceId - > FunctionRuntimeInfo + // NOTE: please use setFunctionRuntimeInfo and deleteFunctionRuntimeInfo methods to modify this data structure + // Since during initialization phase nothing should be modified @VisibleForTesting Map<String, FunctionRuntimeInfo> functionRuntimeInfoMap = new ConcurrentHashMap<>(); @@ -75,7 +80,7 @@ @VisibleForTesting LinkedBlockingQueue<FunctionAction> actionQueue; - private final FunctionAssignmentTailer functionAssignmentTailer; + private FunctionAssignmentTailer functionAssignmentTailer; private FunctionActioner functionActioner; @@ -89,14 +94,16 @@ @Getter private WorkerService workerService; + @Setter + @Getter + boolean isInitializePhase = false; + public FunctionRuntimeManager(WorkerConfig workerConfig, WorkerService workerService, Namespace dlogNamespace, MembershipManager membershipManager, ConnectorsManager connectorsManager) throws Exception { this.workerConfig = workerConfig; this.workerService = workerService; this.functionAdmin = workerService.getFunctionAdmin(); - this.functionAssignmentTailer = new FunctionAssignmentTailer(this); - AuthenticationConfig authConfig = AuthenticationConfig.builder() .clientAuthenticationPlugin(workerConfig.getClientAuthenticationPlugin()) .clientAuthenticationParameters(workerConfig.getClientAuthenticationParameters()) @@ -143,6 +150,41 @@ public FunctionRuntimeManager(WorkerConfig workerConfig, WorkerService workerSer this.membershipManager = membershipManager; } + /** + * Initializes the FunctionRuntimeManager. Does the following: + * 1. Consume all existing assignments to establish existing/latest set of assignments + * 2. After current assignments are read, assignments belonging to this worker will be processed + */ + public void initialize() { + log.info("/** Initializing Runtime Manager **/"); + try { + Reader<byte[]> reader = this.getWorkerService().getClient().newReader() + .topic(this.getWorkerConfig().getFunctionAssignmentTopic()).readCompacted(true) + .startMessageId(MessageId.earliest).create(); + + this.functionAssignmentTailer = new FunctionAssignmentTailer(this, reader); + // read all existing messages + this.setInitializePhase(true); + while (reader.hasMessageAvailable()) { + this.functionAssignmentTailer.processAssignment(reader.readNext()); + } + this.setInitializePhase(false); + // realize existing assignments + Map<String, Assignment> assignmentMap = workerIdToAssignments.get(this.workerConfig.getWorkerId()); + if (assignmentMap != null) { + for (Assignment assignment : assignmentMap.values()) { + startFunctionInstance(assignment); + } + } + // start assignment tailer + this.functionAssignmentTailer.start(); + + } catch (Exception e) { + log.error("Failed to initialize function runtime manager: ", e.getMessage(), e); + throw new RuntimeException(e); + } + } + /** * Starts the function runtime manager */ @@ -622,27 +664,29 @@ void deleteAssignment(Assignment assignment) { } private void addAssignment(Assignment assignment) { - String fullyQualifiedInstanceId = Utils.getFullyQualifiedInstanceId(assignment.getInstance()); - //add new function this.setAssignment(assignment); //Assigned to me if (assignment.getWorkerId().equals(workerConfig.getWorkerId())) { - if (!this.functionRuntimeInfoMap.containsKey(fullyQualifiedInstanceId)) { - this.setFunctionRuntimeInfo(fullyQualifiedInstanceId, new FunctionRuntimeInfo() - .setFunctionInstance(assignment.getInstance())); + startFunctionInstance(assignment); + } + } - } else { - //Somehow this function is already started - log.warn("Function {} already running. Going to restart function.", - this.functionRuntimeInfoMap.get(fullyQualifiedInstanceId)); - this.insertStopAction(this.functionRuntimeInfoMap.get(fullyQualifiedInstanceId)); - } - FunctionRuntimeInfo functionRuntimeInfo = this.functionRuntimeInfoMap.get(fullyQualifiedInstanceId); - this.insertStartAction(functionRuntimeInfo); + private void startFunctionInstance(Assignment assignment) { + String fullyQualifiedInstanceId = Utils.getFullyQualifiedInstanceId(assignment.getInstance()); + if (!this.functionRuntimeInfoMap.containsKey(fullyQualifiedInstanceId)) { + this.setFunctionRuntimeInfo(fullyQualifiedInstanceId, new FunctionRuntimeInfo() + .setFunctionInstance(assignment.getInstance())); + + } else { + //Somehow this function is already started + log.warn("Function {} already running. Going to restart function.", + this.functionRuntimeInfoMap.get(fullyQualifiedInstanceId)); + this.insertStopAction(this.functionRuntimeInfoMap.get(fullyQualifiedInstanceId)); } - + FunctionRuntimeInfo functionRuntimeInfo = this.functionRuntimeInfoMap.get(fullyQualifiedInstanceId); + this.insertStartAction(functionRuntimeInfo); } public Map<String, FunctionRuntimeInfo> getFunctionRuntimeInfos() { @@ -670,26 +714,29 @@ public void updateRates() { @VisibleForTesting void insertStopAction(FunctionRuntimeInfo functionRuntimeInfo) { - FunctionAction functionAction = new FunctionAction(); - functionAction.setAction(FunctionAction.Action.STOP); - functionAction.setFunctionRuntimeInfo(functionRuntimeInfo); - try { - actionQueue.put(functionAction); - } catch (InterruptedException ex) { - throw new RuntimeException("Interrupted while putting action"); + if (!this.isInitializePhase) { + FunctionAction functionAction = new FunctionAction(); + functionAction.setAction(FunctionAction.Action.STOP); + functionAction.setFunctionRuntimeInfo(functionRuntimeInfo); + try { + actionQueue.put(functionAction); + } catch (InterruptedException ex) { + throw new RuntimeException("Interrupted while putting action"); + } } - } @VisibleForTesting void insertStartAction(FunctionRuntimeInfo functionRuntimeInfo) { - FunctionAction functionAction = new FunctionAction(); - functionAction.setAction(FunctionAction.Action.START); - functionAction.setFunctionRuntimeInfo(functionRuntimeInfo); - try { - actionQueue.put(functionAction); - } catch (InterruptedException ex) { - throw new RuntimeException("Interrupted while putting action"); + if (!this.isInitializePhase) { + FunctionAction functionAction = new FunctionAction(); + functionAction.setAction(FunctionAction.Action.START); + functionAction.setFunctionRuntimeInfo(functionRuntimeInfo); + try { + actionQueue.put(functionAction); + } catch (InterruptedException ex) { + throw new RuntimeException("Interrupted while putting action"); + } } } @@ -726,11 +773,16 @@ void setAssignment(Assignment assignment) { } private void deleteFunctionRuntimeInfo(String fullyQualifiedInstanceId) { - this.functionRuntimeInfoMap.remove(fullyQualifiedInstanceId); + if (!this.isInitializePhase) { + this.functionRuntimeInfoMap.remove(fullyQualifiedInstanceId); + } } private void setFunctionRuntimeInfo(String fullyQualifiedInstanceId, FunctionRuntimeInfo functionRuntimeInfo) { - this.functionRuntimeInfoMap.put(fullyQualifiedInstanceId, functionRuntimeInfo); + // Don't modify Function Runtime Infos when initializing + if (!this.isInitializePhase) { + this.functionRuntimeInfoMap.put(fullyQualifiedInstanceId, functionRuntimeInfo); + } } @Override diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java index 488bcd775d..2b9c632479 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java @@ -150,6 +150,9 @@ public void start(URI dlogUri) throws InterruptedException { this.functionRuntimeManager = new FunctionRuntimeManager( this.workerConfig, this, this.dlogNamespace, this.membershipManager, connectorsManager); + // initialize function runtime manager + this.functionRuntimeManager.initialize(); + // Setting references to managers in scheduler this.schedulerManager.setFunctionMetaDataManager(this.functionMetaDataManager); this.schedulerManager.setFunctionRuntimeManager(this.functionRuntimeManager); diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionRuntimeManagerTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionRuntimeManagerTest.java index 5e1ed023aa..490be77732 100644 --- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionRuntimeManagerTest.java +++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionRuntimeManagerTest.java @@ -18,23 +18,31 @@ */ package org.apache.pulsar.functions.worker; +import io.netty.buffer.Unpooled; +import lombok.extern.slf4j.Slf4j; import org.apache.distributedlog.api.namespace.Namespace; import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Reader; import org.apache.pulsar.client.api.ReaderBuilder; +import org.apache.pulsar.client.impl.MessageImpl; +import org.apache.pulsar.functions.metrics.MetricsSink; import org.apache.pulsar.functions.proto.Function; import org.apache.pulsar.functions.proto.InstanceCommunication; -import org.apache.pulsar.functions.proto.Request; -import org.apache.pulsar.functions.metrics.MetricsSink; import org.mockito.ArgumentMatcher; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; import org.testng.Assert; import org.testng.annotations.Test; +import java.util.HashMap; +import java.util.Iterator; import java.util.LinkedList; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyBoolean; @@ -46,7 +54,9 @@ import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +@Slf4j public class FunctionRuntimeManagerTest { public static class TestSink implements MetricsSink { @@ -384,4 +394,122 @@ public boolean matches(Object o) { Assert.assertEquals(functionRuntimeManager.workerIdToAssignments .get("worker-1").get("test-tenant/test-namespace/func-2:0"), assignment3); } + + @Test + public void testRuntimeManagerInitialize() throws Exception { + WorkerConfig workerConfig = new WorkerConfig(); + workerConfig.setWorkerId("worker-1"); + workerConfig.setThreadContainerFactory(new WorkerConfig.ThreadContainerFactory().setThreadGroupName("test")); + workerConfig.setPulsarServiceUrl("pulsar://localhost:6650"); + workerConfig.setStateStorageServiceUrl("foo"); + workerConfig.setFunctionAssignmentTopicName("assignments"); + + Function.FunctionMetaData function1 = Function.FunctionMetaData.newBuilder().setFunctionDetails( + Function.FunctionDetails.newBuilder() + .setTenant("test-tenant").setNamespace("test-namespace").setName("func-1")).build(); + + Function.FunctionMetaData function2 = Function.FunctionMetaData.newBuilder().setFunctionDetails( + Function.FunctionDetails.newBuilder() + .setTenant("test-tenant").setNamespace("test-namespace").setName("func-2")).build(); + + Function.Assignment assignment1 = Function.Assignment.newBuilder() + .setWorkerId("worker-1") + .setInstance(Function.Instance.newBuilder() + .setFunctionMetaData(function1).setInstanceId(0).build()) + .build(); + Function.Assignment assignment2 = Function.Assignment.newBuilder() + .setWorkerId("worker-1") + .setInstance(Function.Instance.newBuilder() + .setFunctionMetaData(function2).setInstanceId(0).build()) + .build(); + + Function.Assignment assignment3 = Function.Assignment.newBuilder() + .setWorkerId("worker-1") + .setInstance(Function.Instance.newBuilder() + .setFunctionMetaData(function2).setInstanceId(0).build()) + .build(); + + List<Message<byte[]>> messageList = new LinkedList<>(); + Message message1 = spy(new MessageImpl("foo", MessageId.latest.toString(), + new HashMap<>(), Unpooled.copiedBuffer(assignment1.toByteArray()), null)); + doReturn(Utils.getFullyQualifiedInstanceId(assignment1.getInstance())).when(message1).getKey(); + + Message message2 = spy(new MessageImpl("foo", MessageId.latest.toString(), + new HashMap<>(), Unpooled.copiedBuffer(assignment2.toByteArray()), null)); + doReturn(Utils.getFullyQualifiedInstanceId(assignment2.getInstance())).when(message2).getKey(); + + // delete function2 + Message message3 = spy(new MessageImpl("foo", MessageId.latest.toString(), + new HashMap<>(), Unpooled.copiedBuffer("".getBytes()), null)); + doReturn(Utils.getFullyQualifiedInstanceId(assignment3.getInstance())).when(message3).getKey(); + + messageList.add(message1); + messageList.add(message2); + messageList.add(message3); + + PulsarClient pulsarClient = mock(PulsarClient.class); + + Reader<byte[]> reader = mock(Reader.class); + + Iterator<Message<byte[]>> it = messageList.iterator(); + + when(reader.readNext()).thenAnswer(new Answer<Message<byte[]>>() { + @Override + public Message<byte[]> answer(InvocationOnMock invocationOnMock) throws Throwable { + return it.next(); + } + }); + + when(reader.readNextAsync()).thenAnswer(new Answer<CompletableFuture<Message<byte[]>>>() { + @Override + public CompletableFuture<Message<byte[]>> answer(InvocationOnMock invocationOnMock) throws Throwable { + return new CompletableFuture<>(); + } + }); + + + when(reader.hasMessageAvailable()).thenAnswer(new Answer<Boolean>() { + @Override + public Boolean answer(InvocationOnMock invocationOnMock) throws Throwable { + return it.hasNext(); + } + }); + + + + ReaderBuilder readerBuilder = mock(ReaderBuilder.class); + doReturn(readerBuilder).when(pulsarClient).newReader(); + doReturn(readerBuilder).when(readerBuilder).topic(anyString()); + doReturn(readerBuilder).when(readerBuilder).startMessageId(any()); + doReturn(readerBuilder).when(readerBuilder).startMessageId(any()); + doReturn(readerBuilder).when(readerBuilder).readCompacted(anyBoolean()); + + doReturn(reader).when(readerBuilder).create(); + WorkerService workerService = mock(WorkerService.class); + doReturn(pulsarClient).when(workerService).getClient(); + doReturn(mock(PulsarAdmin.class)).when(workerService).getFunctionAdmin(); + + // test new assignment add functions + FunctionRuntimeManager functionRuntimeManager = spy(new FunctionRuntimeManager( + workerConfig, + workerService, + mock(Namespace.class), + mock(MembershipManager.class), + mock(ConnectorsManager.class) + )); + + + functionRuntimeManager.initialize(); + + Assert.assertEquals(functionRuntimeManager.workerIdToAssignments.size(), 1); + log.info("actionQueue: {}", functionRuntimeManager.actionQueue); + Assert.assertEquals(functionRuntimeManager.actionQueue.size(), 1); + + FunctionAction functionAction = functionRuntimeManager.actionQueue.poll(); + + // only actually start function1 + Assert.assertEquals(functionAction.getAction(), FunctionAction.Action.START); + Assert.assertEquals(functionAction.getFunctionRuntimeInfo().getFunctionInstance(), assignment1.getInstance()); + + } } ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services