pulsar-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From GitBox <...@apache.org>
Subject [GitHub] rdhabalia closed pull request #2549: [Function] avoid creating assignment snapshot and publish individual assignment msg
Date Tue, 18 Sep 2018 22:22:14 GMT
rdhabalia closed pull request #2549: [Function] avoid creating assignment snapshot and publish individual assignment msg
URL: https://github.com/apache/incubator-pulsar/pull/2549
 
 
   

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/conf/functions_worker.yml b/conf/functions_worker.yml
index 444b7fb9f6..0c7b8af695 100644
--- a/conf/functions_worker.yml
+++ b/conf/functions_worker.yml
@@ -44,6 +44,8 @@ rescheduleTimeoutMs: 60000
 initialBrokerReconnectMaxRetries: 60
 assignmentWriteMaxRetries: 60
 instanceLivenessCheckFreqMs: 30000
+# Frequency how often worker performs compaction on function-topics
+topicCompactionFrequencySec: 1800
 metricsSamplingPeriodSec: 60
 # Enforce authentication
 authenticationEnabled: false
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
index 425e04921b..612b33601b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
@@ -359,4 +359,4 @@ private void phaseTwoLoop(RawReader reader, MessageId to, Map<String, MessageId>
             this.latestForKey = latestForKey;
         }
     }
-}
+}
\ No newline at end of file
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
index 32c93b4626..8e7b292d6a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
@@ -1268,4 +1268,4 @@ public void testEmptyCompactionLedger() throws Exception {
         }
     }
 
-}
+}
\ No newline at end of file
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
new file mode 100644
index 0000000000..fe28a5119e
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
@@ -0,0 +1,370 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.functions.worker;
+
+import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically;
+import static org.mockito.Mockito.spy;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+
+import java.io.File;
+import java.lang.reflect.Method;
+import java.net.InetAddress;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.bookkeeper.test.PortManager;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.ServiceConfigurationUtils;
+import org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl;
+import org.apache.pulsar.client.admin.BrokerStats;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.apache.pulsar.functions.api.utils.IdentityFunction;
+import org.apache.pulsar.functions.proto.Function;
+import org.apache.pulsar.functions.proto.Function.Assignment;
+import org.apache.pulsar.functions.proto.Function.FunctionDetails;
+import org.apache.pulsar.functions.proto.Function.SinkSpec;
+import org.apache.pulsar.functions.proto.Function.SourceSpec;
+import org.apache.pulsar.functions.sink.PulsarSink;
+import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.functions.utils.Utils;
+import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.google.gson.Gson;
+
+import jersey.repackaged.com.google.common.collect.Lists;
+
+/**
+ * Test Pulsar sink on function
+ *
+ */
+public class PulsarWorkerAssignmentTest {
+    LocalBookkeeperEnsemble bkEnsemble;
+
+    ServiceConfiguration config;
+    WorkerConfig workerConfig;
+    PulsarService pulsar;
+    PulsarAdmin admin;
+    PulsarClient pulsarClient;
+    BrokerStats brokerStatsClient;
+    WorkerService functionsWorkerService;
+    final String tenant = "external-repl-prop";
+    String pulsarFunctionsNamespace = tenant + "/use/pulsar-function-admin";
+    String primaryHost;
+    String workerId;
+
+    private final int ZOOKEEPER_PORT = PortManager.nextFreePort();
+    private final int brokerWebServicePort = PortManager.nextFreePort();
+    private final int brokerServicePort = PortManager.nextFreePort();
+    private final int workerServicePort = PortManager.nextFreePort();
+
+    private static final Logger log = LoggerFactory.getLogger(PulsarWorkerAssignmentTest.class);
+
+    @BeforeMethod
+    void setup(Method method) throws Exception {
+
+        log.info("--- Setting up method {} ---", method.getName());
+
+        // Start local bookkeeper ensemble
+        bkEnsemble = new LocalBookkeeperEnsemble(3, ZOOKEEPER_PORT, PortManager.nextFreePort());
+        bkEnsemble.start();
+
+        String brokerServiceUrl = "http://127.0.0.1:" + brokerServicePort;
+        String brokerWeServiceUrl = "http://127.0.0.1:" + brokerWebServicePort;
+
+        config = spy(new ServiceConfiguration());
+        config.setClusterName("use");
+        Set<String> superUsers = Sets.newHashSet("superUser");
+        config.setSuperUserRoles(superUsers);
+        config.setWebServicePort(brokerWebServicePort);
+        config.setZookeeperServers("127.0.0.1" + ":" + ZOOKEEPER_PORT);
+        config.setBrokerServicePort(brokerServicePort);
+        config.setLoadManagerClassName(SimpleLoadManagerImpl.class.getName());
+
+        functionsWorkerService = createPulsarFunctionWorker(config);
+        Optional<WorkerService> functionWorkerService = Optional.of(functionsWorkerService);
+        pulsar = new PulsarService(config, functionWorkerService);
+        pulsar.start();
+
+        admin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerWeServiceUrl).build());
+
+        brokerStatsClient = admin.brokerStats();
+        primaryHost = String.format("http://%s:%d", InetAddress.getLocalHost().getHostName(), brokerWebServicePort);
+
+        // update cluster metadata
+        ClusterData clusterData = new ClusterData(brokerServiceUrl);
+        admin.clusters().updateCluster(config.getClusterName(), clusterData);
+
+        ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(this.workerConfig.getPulsarServiceUrl());
+        pulsarClient = clientBuilder.build();
+
+        TenantInfo propAdmin = new TenantInfo();
+        propAdmin.getAdminRoles().add("superUser");
+        propAdmin.setAllowedClusters(Sets.newHashSet(Lists.newArrayList("use")));
+        admin.tenants().updateTenant(tenant, propAdmin);
+
+        Thread.sleep(100);
+    }
+
+    @AfterMethod
+    void shutdown() throws Exception {
+        log.info("--- Shutting down ---");
+        pulsarClient.close();
+        admin.close();
+        functionsWorkerService.stop();
+        pulsar.close();
+        bkEnsemble.stop();
+    }
+
+    private WorkerService createPulsarFunctionWorker(ServiceConfiguration config) {
+        workerConfig = new WorkerConfig();
+        workerConfig.setPulsarFunctionsNamespace(pulsarFunctionsNamespace);
+        workerConfig.setSchedulerClassName(
+                org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler.class.getName());
+        workerConfig.setThreadContainerFactory(new WorkerConfig.ThreadContainerFactory().setThreadGroupName("use"));
+        // worker talks to local broker
+        workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" + config.getBrokerServicePort());
+        workerConfig.setPulsarWebServiceUrl("http://127.0.0.1:" + config.getWebServicePort());
+        workerConfig.setFailureCheckFreqMs(100);
+        workerConfig.setNumFunctionPackageReplicas(1);
+        workerConfig.setClusterCoordinationTopicName("coordinate");
+        workerConfig.setFunctionAssignmentTopicName("assignment");
+        workerConfig.setFunctionMetadataTopicName("metadata");
+        workerConfig.setInstanceLivenessCheckFreqMs(100);
+        workerConfig.setWorkerPort(workerServicePort);
+        workerConfig.setPulsarFunctionsCluster(config.getClusterName());
+        String hostname = ServiceConfigurationUtils.getDefaultOrConfiguredAddress(config.getAdvertisedAddress());
+        this.workerId = "c-" + config.getClusterName() + "-fw-" + hostname + "-" + workerConfig.getWorkerPort();
+        workerConfig.setWorkerHostname(hostname);
+        workerConfig.setWorkerId(workerId);
+        workerConfig.setTopicCompactionFrequencySec(1);
+
+        return new WorkerService(workerConfig);
+    }
+
+    @Test
+    public void testFunctionAssignments() throws Exception {
+
+        final String namespacePortion = "assignment-test";
+        final String replNamespace = tenant + "/" + namespacePortion;
+        final String sinkTopic = "persistent://" + replNamespace + "/my-topic1";
+        final String functionName = "assign";
+        final String subscriptionName = "test-sub";
+        admin.namespaces().createNamespace(replNamespace);
+        Set<String> clusters = Sets.newHashSet(Lists.newArrayList("use"));
+        admin.namespaces().setNamespaceReplicationClusters(replNamespace, clusters);
+
+        String jarFilePathUrl = Utils.FILE + ":"
+                + PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+        FunctionDetails.Builder functionDetailsBuilder = createFunctionDetails(jarFilePathUrl, tenant, namespacePortion,
+                functionName, "my.*", sinkTopic, subscriptionName);
+        functionDetailsBuilder.setParallelism(2);
+        FunctionDetails functionDetails = functionDetailsBuilder.build();
+
+        // (1) Create function with 2 instance
+        admin.functions().createFunctionWithUrl(functionDetails, jarFilePathUrl);
+        retryStrategically((test) -> {
+            try {
+                return admin.topics().getStats(sinkTopic).subscriptions.size() == 1
+                        && admin.topics().getStats(sinkTopic).subscriptions.values().iterator().next().consumers
+                                .size() == 2;
+            } catch (PulsarAdminException e) {
+                return false;
+            }
+        }, 5, 150);
+        // validate 2 instances have been started
+        assertEquals(admin.topics().getStats(sinkTopic).subscriptions.size(), 1);
+        assertEquals(admin.topics().getStats(sinkTopic).subscriptions.values().iterator().next().consumers.size(), 2);
+
+        // (2) Update function with 1 instance
+        functionDetailsBuilder.setParallelism(1);
+        functionDetails = functionDetailsBuilder.build();
+        // try to update function to test: update-function functionality
+        admin.functions().updateFunctionWithUrl(functionDetails, jarFilePathUrl);
+        retryStrategically((test) -> {
+            try {
+                return admin.topics().getStats(sinkTopic).subscriptions.size() == 1
+                        && admin.topics().getStats(sinkTopic).subscriptions.values().iterator().next().consumers
+                                .size() == 1;
+            } catch (PulsarAdminException e) {
+                return false;
+            }
+        }, 5, 150);
+        // validate pulsar sink consumer has started on the topic
+        assertEquals(admin.topics().getStats(sinkTopic).subscriptions.values().iterator().next().consumers.size(), 1);
+    }
+
+    @Test(timeOut=20000)
+    public void testFunctionAssignmentsWithRestart() throws Exception {
+
+        final String namespacePortion = "assignment-test";
+        final String replNamespace = tenant + "/" + namespacePortion;
+        final String sinkTopic = "persistent://" + replNamespace + "/my-topic1";
+        final String baseFunctionName = "assign-restart";
+        final String subscriptionName = "test-sub";
+        final int totalFunctions = 5;
+        final int parallelism = 2;
+        admin.namespaces().createNamespace(replNamespace);
+        Set<String> clusters = Sets.newHashSet(Lists.newArrayList("use"));
+        admin.namespaces().setNamespaceReplicationClusters(replNamespace, clusters);
+        final FunctionRuntimeManager runtimeManager = functionsWorkerService.getFunctionRuntimeManager();
+
+        String jarFilePathUrl = Utils.FILE + ":"
+                + PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+        FunctionDetails.Builder functionDetailsBuilder = null;
+        // (1) Register functions with 2 instances
+        for (int i = 0; i < totalFunctions; i++) {
+            String functionName = baseFunctionName + i;
+            functionDetailsBuilder = createFunctionDetails(jarFilePathUrl, tenant, namespacePortion, functionName,
+                    "my.*", sinkTopic, subscriptionName);
+            functionDetailsBuilder.setParallelism(parallelism);
+            // set-auto-ack prop =true
+            functionDetailsBuilder.setAutoAck(true);
+            FunctionDetails functionDetails = functionDetailsBuilder.build();
+            admin.functions().createFunctionWithUrl(functionDetails, jarFilePathUrl);
+        }
+        retryStrategically((test) -> {
+            try {
+                Map<String, Assignment> assgn = runtimeManager.getCurrentAssignments().values().iterator().next();
+                return assgn.size() == (totalFunctions * parallelism);
+            } catch (Exception e) {
+                return false;
+            }
+        }, 5, 150);
+
+        // Validate registered assignments
+        Map<String, Assignment> assignments = runtimeManager.getCurrentAssignments().values().iterator().next();
+        assertEquals(assignments.size(), (totalFunctions * parallelism));
+
+        // (2) Update function with prop=auto-ack and Delete 2 functions
+        for (int i = 0; i < totalFunctions; i++) {
+            String functionName = baseFunctionName + i;
+            functionDetailsBuilder = createFunctionDetails(jarFilePathUrl, tenant, namespacePortion, functionName,
+                    "my.*", sinkTopic, subscriptionName);
+            functionDetailsBuilder.setParallelism(parallelism);
+            // set-auto-ack prop =false
+            functionDetailsBuilder.setAutoAck(false);
+            FunctionDetails functionDetails = functionDetailsBuilder.build();
+            admin.functions().updateFunctionWithUrl(functionDetails, jarFilePathUrl);
+        }
+
+        int totalDeletedFunction = 2;
+        for (int i = (totalFunctions - 1); i >= (totalFunctions - totalDeletedFunction); i--) {
+            String functionName = baseFunctionName + i;
+            admin.functions().deleteFunction(tenant, namespacePortion, functionName);
+        }
+        retryStrategically((test) -> {
+            try {
+                Map<String, Assignment> assgn = runtimeManager.getCurrentAssignments().values().iterator().next();
+                return assgn.size() == ((totalFunctions - totalDeletedFunction) * parallelism);
+            } catch (Exception e) {
+                return false;
+            }
+        }, 5, 150);
+
+        // Validate registered assignments
+        assignments = runtimeManager.getCurrentAssignments().values().iterator().next();
+        assertEquals(assignments.size(), ((totalFunctions - totalDeletedFunction) * parallelism));
+
+        // (3) Restart worker service and check registered functions
+        URI dlUri = functionsWorkerService.getDlogUri();
+        functionsWorkerService.stop();
+        functionsWorkerService = new WorkerService(workerConfig);
+        functionsWorkerService.start(dlUri);
+        FunctionRuntimeManager runtimeManager2 = functionsWorkerService.getFunctionRuntimeManager();
+        retryStrategically((test) -> {
+            try {
+                Map<String, Assignment> assgn = runtimeManager2.getCurrentAssignments().values().iterator().next();
+                return assgn.size() == ((totalFunctions - totalDeletedFunction) * parallelism);
+            } catch (Exception e) {
+                return false;
+            }
+        }, 5, 150);
+
+        // Validate registered assignments
+        assignments = runtimeManager2.getCurrentAssignments().values().iterator().next();
+        assertEquals(assignments.size(), ((totalFunctions - totalDeletedFunction) * parallelism));
+
+        // validate updated function prop = auto-ack=false and instnaceid
+        for (int i = 0; i < (totalFunctions - totalDeletedFunction); i++) {
+            String functionName = baseFunctionName + i;
+            assertFalse(admin.functions().getFunction(tenant, namespacePortion, functionName).getAutoAck());
+        }
+    }
+
+    protected static FunctionDetails.Builder createFunctionDetails(String jarFile, String tenant, String namespace,
+            String functionName, String sourceTopic, String sinkTopic, String subscriptionName) {
+
+        File file = new File(jarFile);
+        try {
+            Reflections.loadJar(file);
+        } catch (MalformedURLException e) {
+            throw new RuntimeException("Failed to load user jar " + file, e);
+        }
+        String sourceTopicPattern = String.format("persistent://%s/%s/%s", tenant, namespace, sourceTopic);
+        Class<?> typeArg = byte[].class;
+
+        FunctionDetails.Builder functionDetailsBuilder = FunctionDetails.newBuilder();
+        functionDetailsBuilder.setTenant(tenant);
+        functionDetailsBuilder.setNamespace(namespace);
+        functionDetailsBuilder.setName(functionName);
+        functionDetailsBuilder.setRuntime(FunctionDetails.Runtime.JAVA);
+        functionDetailsBuilder.setParallelism(1);
+        functionDetailsBuilder.setClassName(IdentityFunction.class.getName());
+
+        // set source spec
+        // source spec classname should be empty so that the default pulsar source will be used
+        SourceSpec.Builder sourceSpecBuilder = SourceSpec.newBuilder();
+        sourceSpecBuilder.setSubscriptionType(Function.SubscriptionType.SHARED);
+        sourceSpecBuilder.setTypeClassName(typeArg.getName());
+        sourceSpecBuilder.setTopicsPattern(sourceTopicPattern);
+        sourceSpecBuilder.setSubscriptionName(subscriptionName);
+        sourceSpecBuilder.putTopicsToSerDeClassName(sourceTopicPattern, "");
+        functionDetailsBuilder.setAutoAck(true);
+        functionDetailsBuilder.setSource(sourceSpecBuilder);
+
+        // set up sink spec
+        SinkSpec.Builder sinkSpecBuilder = SinkSpec.newBuilder();
+        // sinkSpecBuilder.setClassName(PulsarSink.class.getName());
+        sinkSpecBuilder.setTopic(sinkTopic);
+        Map<String, Object> sinkConfigMap = Maps.newHashMap();
+        sinkSpecBuilder.setConfigs(new Gson().toJson(sinkConfigMap));
+        sinkSpecBuilder.setTypeClassName(typeArg.getName());
+        functionDetailsBuilder.setSink(sinkSpecBuilder);
+
+        return functionDetailsBuilder;
+    }
+
+}
\ No newline at end of file
diff --git a/pulsar-functions/proto/src/main/proto/Request.proto b/pulsar-functions/proto/src/main/proto/Request.proto
index c8e31d75ed..574346911c 100644
--- a/pulsar-functions/proto/src/main/proto/Request.proto
+++ b/pulsar-functions/proto/src/main/proto/Request.proto
@@ -37,8 +37,3 @@ message ServiceRequest {
     FunctionMetaData functionMetaData = 3;
     string workerId = 4;
 }
-
-message AssignmentsUpdate {
-    repeated Assignment assignments = 1;
-    uint64 version = 2;
-}
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 366eaba6a0..3ad6c7c6ee 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
@@ -18,14 +18,16 @@
  */
 package org.apache.pulsar.functions.worker;
 
-import lombok.extern.slf4j.Slf4j;
+import java.io.IOException;
+import java.util.function.Function;
+
 import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Reader;
-import org.apache.pulsar.functions.proto.Request;
+import org.apache.pulsar.functions.proto.Function.Assignment;
 
-import java.io.IOException;
-import java.util.function.Function;
+import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
 public class FunctionAssignmentTailer
@@ -34,15 +36,16 @@
         private final FunctionRuntimeManager functionRuntimeManager;
         private final Reader<byte[]> reader;
 
-    public FunctionAssignmentTailer(FunctionRuntimeManager functionRuntimeManager,
-                Reader<byte[]> reader)
+    public FunctionAssignmentTailer(FunctionRuntimeManager functionRuntimeManager)
             throws PulsarClientException {
-            this.functionRuntimeManager = functionRuntimeManager;
-            this.reader = reader;
-        }
+        this.functionRuntimeManager = functionRuntimeManager;
 
-    public void start() {
+        this.reader = functionRuntimeManager.getWorkerService().getClient().newReader()
+                .topic(functionRuntimeManager.getWorkerConfig().getFunctionAssignmentTopic()).readCompacted(true)
+                .startMessageId(MessageId.earliest).create();
+    }
 
+    public void start() {
         receiveOne();
     }
 
@@ -65,29 +68,21 @@ public void close() {
 
     @Override
     public void accept(Message<byte[]> msg) {
-
-        // check if latest
-        boolean hasMessageAvailable;
-        try {
-            hasMessageAvailable = this.reader.hasMessageAvailable();
-        } catch (PulsarClientException e) {
-            throw new RuntimeException(e);
-        }
-        if (!hasMessageAvailable) {
-            Request.AssignmentsUpdate assignmentsUpdate;
+        if(msg.getData()==null || (msg.getData().length==0)) {
+            log.info("Received assignment delete: {}", msg.getKey());
+            this.functionRuntimeManager.deleteAssignment(msg.getKey());
+        } else {
+            Assignment assignment;
             try {
-                assignmentsUpdate = Request.AssignmentsUpdate.parseFrom(msg.getData());
+                assignment = Assignment.parseFrom(msg.getData());
             } catch (IOException e) {
                 log.error("[{}] Received bad assignment update at message {}", reader.getTopic(), msg.getMessageId(),
                         e);
                 // TODO: find a better way to handle bad request
                 throw new RuntimeException(e);
             }
-            if (log.isDebugEnabled()) {
-                log.debug("Received assignment update: {}", assignmentsUpdate);
-            }
-
-            this.functionRuntimeManager.processAssignmentUpdate(msg.getMessageId(), assignmentsUpdate);
+            log.info("Received assignment update: {}", assignment);
+            this.functionRuntimeManager.processAssignment(assignment);    
         }
         // receive next request
         receiveOne();
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 43cd27bcb5..e634130f5e 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
@@ -18,48 +18,45 @@
  */
 package org.apache.pulsar.functions.worker;
 
-import com.google.common.annotations.VisibleForTesting;
 import java.io.IOException;
 import java.net.URI;
-import java.net.URISyntaxException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.stream.Collectors;
+
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.client.Client;
+import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
+import javax.ws.rs.core.UriBuilder;
 
-import lombok.extern.slf4j.Slf4j;
 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.PulsarClient;
-import org.apache.pulsar.client.api.Reader;
 import org.apache.pulsar.common.policies.data.ErrorData;
-import org.apache.pulsar.functions.proto.Function.Assignment;
 import org.apache.pulsar.functions.instance.AuthenticationConfig;
+import org.apache.pulsar.functions.proto.Function.Assignment;
 import org.apache.pulsar.functions.proto.InstanceCommunication;
-import org.apache.pulsar.functions.proto.Request.AssignmentsUpdate;
-import org.apache.pulsar.functions.runtime.RuntimeFactory;
 import org.apache.pulsar.functions.runtime.ProcessRuntimeFactory;
 import org.apache.pulsar.functions.runtime.Runtime;
-import org.apache.pulsar.functions.runtime.ThreadRuntimeFactory;
+import org.apache.pulsar.functions.runtime.RuntimeFactory;
 import org.apache.pulsar.functions.runtime.RuntimeSpawner;
+import org.apache.pulsar.functions.runtime.ThreadRuntimeFactory;
 
-import javax.ws.rs.WebApplicationException;
-import javax.ws.rs.client.Client;
-import javax.ws.rs.client.ClientBuilder;
-import javax.ws.rs.client.Entity;
-import javax.ws.rs.core.MediaType;
-import javax.ws.rs.core.Response;
-import javax.ws.rs.core.UriBuilder;
-import javax.ws.rs.core.Response.Status;
+import com.google.common.annotations.VisibleForTesting;
 
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.stream.Collectors;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
 
 /**
  * This class managers all aspects of functions assignments and running of function assignments for this worker
@@ -78,13 +75,12 @@
     Map<String, FunctionRuntimeInfo> functionRuntimeInfoMap = new ConcurrentHashMap<>();
 
     @VisibleForTesting
+    @Getter
     final WorkerConfig workerConfig;
 
     @VisibleForTesting
     LinkedBlockingQueue<FunctionAction> actionQueue;
 
-    private long currentAssignmentVersion = 0;
-
     private final FunctionAssignmentTailer functionAssignmentTailer;
 
     private FunctionActioner functionActioner;
@@ -92,22 +88,19 @@
     private RuntimeFactory runtimeFactory;
 
     private MembershipManager membershipManager;
-    private final ConnectorsManager connectorsManager;
     
     private final PulsarAdmin functionAdmin;
+    
+    @Getter
+    private WorkerService workerService;
 
     public FunctionRuntimeManager(WorkerConfig workerConfig, WorkerService workerService, Namespace dlogNamespace,
             MembershipManager membershipManager, ConnectorsManager connectorsManager) throws Exception {
         this.workerConfig = workerConfig;
-        this.connectorsManager = connectorsManager;
+        this.workerService = workerService;
         this.functionAdmin = workerService.getFunctionAdmin();
 
-        Reader<byte[]> reader = workerService.getClient().newReader()
-                .topic(this.workerConfig.getFunctionAssignmentTopic())
-                .startMessageId(MessageId.earliest)
-                .create();
-
-        this.functionAssignmentTailer = new FunctionAssignmentTailer(this, reader);
+        this.functionAssignmentTailer = new FunctionAssignmentTailer(this);
 
         AuthenticationConfig authConfig = AuthenticationConfig.builder()
                 .clientAuthenticationPlugin(workerConfig.getClientAuthenticationPlugin())
@@ -222,13 +215,6 @@ public synchronized Assignment findFunctionAssignment(String tenant, String name
         return assignments;
     }
 
-    /**
-     * get the current version number of assignments
-     * @return assignments version number
-     */
-    public synchronized long getCurrentAssignmentVersion() {
-        return new Long(this.currentAssignmentVersion);
-    }
 
     /**
      * Removes a collection of assignments
@@ -461,105 +447,108 @@ private void stopFunction(String fullyQualifiedInstanceId, boolean restart) thro
     /**
      * Process an assignment update from the assignment topic
      * @param messageId the message id of the update assignment
-     * @param assignmentsUpdate the assignment update
+     * @param newAssignment the assignment
      */
-    public synchronized void processAssignmentUpdate(MessageId messageId, AssignmentsUpdate assignmentsUpdate) {
+    public synchronized void processAssignment(Assignment newAssignment) {
 
-        if (assignmentsUpdate.getVersion() > this.currentAssignmentVersion) {
+        Map<String, Assignment> existingAssignmentMap = new HashMap<>();
+        for (Map<String, Assignment> entry : this.workerIdToAssignments.values()) {
+            existingAssignmentMap.putAll(entry);
+        }
 
-            Map<String, Assignment> assignmentMap = new HashMap<>();
-            for (Assignment assignment : assignmentsUpdate.getAssignmentsList()) {
-                assignmentMap.put(
-                        Utils.getFullyQualifiedInstanceId(assignment.getInstance()),
-                        assignment);
+        if (existingAssignmentMap.containsKey(Utils.getFullyQualifiedInstanceId(newAssignment.getInstance()))) {
+            updateAssignment(newAssignment);
+        } else {
+            addAssignment(newAssignment);
+        }
+    }
+
+    private void updateAssignment(Assignment assignment) {
+        String fullyQualifiedInstanceId = Utils.getFullyQualifiedInstanceId(assignment.getInstance());
+        Assignment existingAssignment = this.findAssignment(assignment);
+        // potential updates need to happen
+        if (!existingAssignment.equals(assignment)) {
+            FunctionRuntimeInfo functionRuntimeInfo = this.functionRuntimeInfoMap.get(fullyQualifiedInstanceId);
+            //stop function
+            if (functionRuntimeInfo != null) {
+                this.insertStopAction(functionRuntimeInfo);
             }
-            Map<String, Assignment> existingAssignmentMap = new HashMap<>();
-            for (Map<String, Assignment> entry : this.workerIdToAssignments.values()) {
-                existingAssignmentMap.putAll(entry);
+            // still assigned to me, need to restart
+            if (assignment.getWorkerId().equals(this.workerConfig.getWorkerId())) {
+                //start again
+                FunctionRuntimeInfo newFunctionRuntimeInfo = new FunctionRuntimeInfo();
+                newFunctionRuntimeInfo.setFunctionInstance(assignment.getInstance());
+                this.insertStartAction(newFunctionRuntimeInfo);
+                this.setFunctionRuntimeInfo(fullyQualifiedInstanceId, newFunctionRuntimeInfo);
             }
 
-            Map<String, Assignment> assignmentsToAdd = diff(assignmentMap, existingAssignmentMap);
-
-            Map<String, Assignment> assignmentsToDelete = diff(existingAssignmentMap, assignmentMap);
-
-            Map<String, Assignment> existingAssignments = inCommon(assignmentMap, existingAssignmentMap);
-
-            // functions to add
-            for (Map.Entry<String, Assignment> assignmentEntry : assignmentsToAdd.entrySet()) {
-                String fullyQualifiedInstanceId = assignmentEntry.getKey();
-                Assignment assignment = assignmentEntry.getValue();
-
-                //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()));
-
-                    } 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);
-                }
+            // find existing assignment
+            Assignment existing_assignment = this.findAssignment(assignment);
+            if (existing_assignment != null) {
+                // delete old assignment that could have old data
+                this.deleteAssignment(existing_assignment);
             }
+            // set to newest assignment
+            this.setAssignment(assignment);
+        }
+    }
+   
+    public synchronized void deleteAssignment(String fullyQualifiedInstanceId) {
+        FunctionRuntimeInfo functionRuntimeInfo = this.functionRuntimeInfoMap.get(fullyQualifiedInstanceId);
+        if (functionRuntimeInfo != null) {
+            this.insertStopAction(functionRuntimeInfo);
+            this.deleteFunctionRuntimeInfo(fullyQualifiedInstanceId);
+        }
+        
+        String workerId = null;
+        for(Entry<String, Map<String, Assignment>> workerAssignments : workerIdToAssignments.entrySet()) {
+            if(workerAssignments.getValue().remove(fullyQualifiedInstanceId)!=null) {
+                workerId = workerAssignments.getKey();
+                break;
+            }
+        }
+        Map<String, Assignment> worker;
+        if (workerId != null && ((worker = workerIdToAssignments.get(workerId)) != null && worker.isEmpty())) {
+            this.workerIdToAssignments.remove(workerId);
+        }
+    }
 
-            // functions to delete
-            for (Map.Entry<String, Assignment> assignmentEntry : assignmentsToDelete.entrySet()) {
-                String fullyQualifiedInstanceId = assignmentEntry.getKey();
-                Assignment assignment = assignmentEntry.getValue();
-
-                FunctionRuntimeInfo functionRuntimeInfo = this.functionRuntimeInfoMap.get(fullyQualifiedInstanceId);
-                if (functionRuntimeInfo != null) {
-                    this.insertStopAction(functionRuntimeInfo);
-                    this.deleteFunctionRuntimeInfo(fullyQualifiedInstanceId);
-                }
-                this.deleteAssignment(assignment);
+    @VisibleForTesting
+    void deleteAssignment(Assignment assignment) {
+        String fullyQualifiedInstanceId = Utils.getFullyQualifiedInstanceId(assignment.getInstance());
+        Map<String, Assignment> assignmentMap = this.workerIdToAssignments.get(assignment.getWorkerId());
+        if (assignmentMap != null) {
+            if (assignmentMap.containsKey(fullyQualifiedInstanceId)) {
+                assignmentMap.remove(fullyQualifiedInstanceId);
             }
+            if (assignmentMap.isEmpty()) {
+                this.workerIdToAssignments.remove(assignment.getWorkerId());
+            }
+        }
+    }
 
-            // functions to update
-            for (Map.Entry<String, Assignment> assignmentEntry : existingAssignments.entrySet()) {
-                String fullyQualifiedInstanceId = assignmentEntry.getKey();
-                Assignment assignment = assignmentEntry.getValue();
-                Assignment existingAssignment = this.findAssignment(assignment);
-                // potential updates need to happen
-                if (!existingAssignment.equals(assignment)) {
-                    FunctionRuntimeInfo functionRuntimeInfo = this.functionRuntimeInfoMap.get(fullyQualifiedInstanceId);
-                    //stop function
-                    if (functionRuntimeInfo != null) {
-                        this.insertStopAction(functionRuntimeInfo);
-                    }
-                    // still assigned to me, need to restart
-                    if (assignment.getWorkerId().equals(this.workerConfig.getWorkerId())) {
-                        //start again
-                        FunctionRuntimeInfo newFunctionRuntimeInfo = new FunctionRuntimeInfo();
-                        newFunctionRuntimeInfo.setFunctionInstance(assignment.getInstance());
-                        this.insertStartAction(newFunctionRuntimeInfo);
-                        this.setFunctionRuntimeInfo(fullyQualifiedInstanceId, newFunctionRuntimeInfo);
-                    }
+    private void addAssignment(Assignment assignment) {
+        String fullyQualifiedInstanceId = Utils.getFullyQualifiedInstanceId(assignment.getInstance());
 
-                    // find existing assignment
-                    Assignment existing_assignment = this.findAssignment(assignment);
-                    if (existing_assignment != null) {
-                        // delete old assignment that could have old data
-                        this.deleteAssignment(existing_assignment);
-                    }
-                    // set to newest assignment
-                    this.setAssignment(assignment);
-                }
-            }
+        //add new function
+        this.setAssignment(assignment);
 
-            // set as current assignment
-            this.currentAssignmentVersion = assignmentsUpdate.getVersion();
+        //Assigned to me
+        if (assignment.getWorkerId().equals(workerConfig.getWorkerId())) {
+            if (!this.functionRuntimeInfoMap.containsKey(fullyQualifiedInstanceId)) {
+                this.setFunctionRuntimeInfo(fullyQualifiedInstanceId, new FunctionRuntimeInfo()
+                        .setFunctionInstance(assignment.getInstance()));
 
-        } else {
-            log.debug("Received out of date assignment update: {}", assignmentsUpdate);
+            } 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() {
@@ -642,20 +631,6 @@ void setAssignment(Assignment assignment) {
                 assignment);
     }
 
-    @VisibleForTesting
-    void deleteAssignment(Assignment assignment) {
-        Map<String, Assignment> assignmentMap = this.workerIdToAssignments.get(assignment.getWorkerId());
-        if (assignmentMap != null) {
-            String fullyQualifiedInstanceId = Utils.getFullyQualifiedInstanceId(assignment.getInstance());
-            if (assignmentMap.containsKey(fullyQualifiedInstanceId)) {
-                assignmentMap.remove(fullyQualifiedInstanceId);
-            }
-            if (assignmentMap.isEmpty()) {
-                this.workerIdToAssignments.remove(assignment.getWorkerId());
-            }
-        }
-    }
-
     private void deleteFunctionRuntimeInfo(String fullyQualifiedInstanceId) {
         this.functionRuntimeInfoMap.remove(fullyQualifiedInstanceId);
     }
@@ -674,28 +649,8 @@ public void close() throws Exception {
         }
     }
 
-    private Map<String, Assignment> diff(Map<String, Assignment> assignmentMap1, Map<String, Assignment> assignmentMap2) {
-        Map<String, Assignment> result = new HashMap<>();
-        for (Map.Entry<String, Assignment> entry : assignmentMap1.entrySet()) {
-            if (!assignmentMap2.containsKey(entry.getKey())) {
-                result.put(entry.getKey(), entry.getValue());
-            }
-        }
-        return result;
-    }
-
-    private Map<String, Assignment> inCommon(Map<String, Assignment> assignmentMap1, Map<String, Assignment> assignmentMap2) {
-
-        Map<String, Assignment> result = new HashMap<>();
-        for (Map.Entry<String, Assignment> entry : assignmentMap1.entrySet()) {
-            if (assignmentMap2.containsKey(entry.getKey())) {
-                result.put(entry.getKey(), entry.getValue());
-            }
-        }
-        return result;
-    }
-
     private FunctionRuntimeInfo getFunctionRuntimeInfo(String fullyQualifiedInstanceId) {
         return this.functionRuntimeInfoMap.get(fullyQualifiedInstanceId);
     }
+
 }
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
index ed00958d7e..db7785a55b 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
@@ -23,30 +23,29 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
-import lombok.Setter;
-import lombok.extern.slf4j.Slf4j;
-
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.CompressionType;
-import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.functions.proto.Function;
 import org.apache.pulsar.functions.proto.Function.Assignment;
 import org.apache.pulsar.functions.proto.Function.FunctionMetaData;
-import org.apache.pulsar.functions.proto.Request;
 import org.apache.pulsar.functions.utils.Reflections;
 import org.apache.pulsar.functions.worker.scheduler.IScheduler;
 
+import com.google.common.annotations.VisibleForTesting;
+
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+
 @Slf4j
 public class SchedulerManager implements AutoCloseable {
 
@@ -65,16 +64,22 @@
 
     private final Producer<byte[]> producer;
 
-    private final ExecutorService executorService;
+    private final ScheduledExecutorService executorService;
+    
+    private final PulsarAdmin admin;
+    
+    AtomicBoolean isCompactionNeeded = new AtomicBoolean(false);
+    private static final long DEFAULT_ADMIN_API_BACKOFF_SEC = 60; 
 
-    public SchedulerManager(WorkerConfig workerConfig, PulsarClient pulsarClient) {
+    public SchedulerManager(WorkerConfig workerConfig, PulsarClient pulsarClient, PulsarAdmin admin, ScheduledExecutorService executor) {
         this.workerConfig = workerConfig;
+        this.admin = admin;
         this.scheduler = Reflections.createInstance(workerConfig.getSchedulerClassName(), IScheduler.class,
                 Thread.currentThread().getContextClassLoader());
 
         try {
             this.producer = pulsarClient.newProducer().topic(this.workerConfig.getFunctionAssignmentTopic())
-                    .enableBatching(true).blockIfQueueFull(true).compressionType(CompressionType.LZ4).
+                    .enableBatching(false).blockIfQueueFull(true).compressionType(CompressionType.LZ4).
                     sendTimeout(0, TimeUnit.MILLISECONDS).create();
         } catch (PulsarClientException e) {
             log.error("Failed to create producer to function assignment topic "
@@ -82,9 +87,9 @@ public SchedulerManager(WorkerConfig workerConfig, PulsarClient pulsarClient) {
             throw new RuntimeException(e);
         }
 
-        this.executorService =
-                new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
-                        new LinkedBlockingQueue<>());
+        this.executorService = executor;
+        
+        scheduleCompaction(executor, workerConfig.getTopicCompactionFrequencySec());
     }
 
     public Future<?> schedule() {
@@ -92,13 +97,31 @@ public SchedulerManager(WorkerConfig workerConfig, PulsarClient pulsarClient) {
             synchronized (SchedulerManager.this) {
                 boolean isLeader = membershipManager.isLeader();
                 if (isLeader) {
-                    invokeScheduler();
+                    try {
+                        invokeScheduler();
+                    } catch (Exception e) {
+                        log.warn("Failed to invoke scheduler", e);
+                        schedule();
+                    }
                 }
             }
         });
     }
 
-    private void invokeScheduler() {
+    private void scheduleCompaction(ScheduledExecutorService executor, long scheduleFrequencySec) {
+        if (executor != null) {
+            executor.scheduleWithFixedDelay(() -> {
+                if (membershipManager.isLeader() && isCompactionNeeded.get()) {
+                    compactAssignmentTopic();
+                    isCompactionNeeded.set(false);
+                }
+            }, scheduleFrequencySec, scheduleFrequencySec, TimeUnit.SECONDS);
+        }
+    }
+    
+    @VisibleForTesting
+    public void invokeScheduler() {
+        
         List<String> currentMembership = this.membershipManager.getCurrentMembership()
                 .stream().map(workerInfo -> workerInfo.getWorkerId()).collect(Collectors.toList());
 
@@ -111,12 +134,16 @@ private void invokeScheduler() {
         while (it.hasNext()) {
             Map.Entry<String, Map<String, Assignment>> workerIdToAssignmentEntry = it.next();
             Map<String, Assignment> functionMap = workerIdToAssignmentEntry.getValue();
+
             // remove instances that don't exist anymore
-            functionMap.entrySet().removeIf(
-                    entry -> {
-                        String fullyQualifiedInstanceId = entry.getKey();
-                        return !allInstances.containsKey(fullyQualifiedInstanceId);
-                    });
+            functionMap.entrySet().removeIf(entry -> {
+                String fullyQualifiedInstanceId = entry.getKey();
+                boolean deleted = !allInstances.containsKey(fullyQualifiedInstanceId);
+                if (deleted) {
+                    publishNewAssignment(entry.getValue().toBuilder().build(), true);
+                }
+                return deleted;
+            });
 
             // update assignment instances in case attributes of a function gets updated
             for (Map.Entry<String, Assignment> entry : functionMap.entrySet()) {
@@ -143,36 +170,40 @@ private void invokeScheduler() {
         List<Assignment> assignments = this.scheduler.schedule(
                 needsAssignment, currentAssignments, currentMembership);
 
-        log.debug("New assignments computed: {}", assignments);
+        if (log.isDebugEnabled()) {
+            log.debug("New assignments computed: {}", assignments);
+        }
 
-        long assignmentVersion = this.functionRuntimeManager.getCurrentAssignmentVersion() + 1;
-        Request.AssignmentsUpdate assignmentsUpdate = Request.AssignmentsUpdate.newBuilder()
-                .setVersion(assignmentVersion)
-                .addAllAssignments(assignments)
-                .build();
+        isCompactionNeeded.set(!assignments.isEmpty());
 
-        CompletableFuture<MessageId> messageIdCompletableFuture = producer.sendAsync(assignmentsUpdate.toByteArray());
-        try {
-            messageIdCompletableFuture.get();
-        } catch (InterruptedException | ExecutionException e) {
-            log.error("Failed to send assignment update", e);
-            throw new RuntimeException(e);
+        for(Assignment assignment : assignments) {
+            publishNewAssignment(assignment, false);
         }
+        
+    }
 
-        // wait for assignment update to go throw the pipeline
-        int retries = 0;
-        while (this.functionRuntimeManager.getCurrentAssignmentVersion() < assignmentVersion) {
-            if (retries >= this.workerConfig.getAssignmentWriteMaxRetries()) {
-                log.warn("Max number of retries reached for waiting for assignment to propagate. Will continue now.");
-                break;
-            }
-            log.info("Waiting for assignments to propagate...");
+    public void compactAssignmentTopic() {
+        if (this.admin != null) {
             try {
-                Thread.sleep(500);
-            } catch (InterruptedException e) {
-                throw new RuntimeException(e);
+                this.admin.topics().triggerCompaction(workerConfig.getFunctionAssignmentTopic());
+            } catch (PulsarAdminException e) {
+                log.error("Failed to trigger compaction {}", e);
+                executorService.schedule(() -> compactAssignmentTopic(), DEFAULT_ADMIN_API_BACKOFF_SEC,
+                        TimeUnit.SECONDS);
             }
-            retries++;
+        }
+    }
+
+    private void publishNewAssignment(Assignment assignment, boolean deleted) {
+        try {
+            String fullyQualifiedInstanceId = Utils.getFullyQualifiedInstanceId(assignment.getInstance());
+            // publish empty message with instance-id key so, compactor can delete and skip delivery of this instance-id
+            // message 
+            producer.newMessage().key(fullyQualifiedInstanceId)
+                    .value(deleted ? "".getBytes() : assignment.toByteArray()).sendAsync().get();
+        } catch (Exception e) {
+            log.error("Failed to {} assignment update {}", assignment, deleted ? "send" : "deleted", e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -224,6 +255,5 @@ public void close() {
         } catch (PulsarClientException e) {
             log.warn("Failed to shutdown scheduler manager assignment producer", e);
         }
-        this.executorService.shutdown();
     }
 }
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
index 0f695a974f..a2524c6d97 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
@@ -74,6 +74,9 @@
     private long instanceLivenessCheckFreqMs;
     private String clientAuthenticationPlugin;
     private String clientAuthenticationParameters;
+    // Frequency how often worker performs compaction on function-topics
+    private long topicCompactionFrequencySec = 30 * 60; // 30 minutes
+    private int metricsSamplingPeriodSec = 60;
     /***** --- TLS --- ****/
     // Enable TLS
     private boolean tlsEnabled = false;
@@ -88,8 +91,6 @@
     private boolean tlsRequireTrustedClientCertOnConnect = false;
     private boolean useTls = false;
     private boolean tlsHostnameVerificationEnable = false;
-    
-    private int metricsSamplingPeriodSec = 60;
     // Enforce authentication
     private boolean authenticationEnabled = false;
     // Autentication provider name list, which is a list of class names
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 7fc0cc94d7..488bcd775d 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
@@ -20,6 +20,7 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.annotations.VisibleForTesting;
 
 import io.netty.util.concurrent.DefaultThreadFactory;
 
@@ -68,11 +69,15 @@
     private PulsarAdmin brokerAdmin;
     private PulsarAdmin functionAdmin;
     private final MetricsGenerator metricsGenerator;
+    private final ScheduledExecutorService executor;
+    @VisibleForTesting
+    private URI dlogUri;
 
     public WorkerService(WorkerConfig workerConfig) {
         this.workerConfig = workerConfig;
         this.statsUpdater = Executors
                 .newSingleThreadScheduledExecutor(new DefaultThreadFactory("worker-stats-updater"));
+        this.executor = Executors.newScheduledThreadPool(10, new DefaultThreadFactory("pulsar-worker"));
         this.metricsGenerator = new MetricsGenerator(this.statsUpdater);
     }
 
@@ -98,12 +103,13 @@ public void start(URI dlogUri) throws InterruptedException {
         }
 
         // create the dlog namespace for storing function packages
+        this.dlogUri = dlogUri;
         DistributedLogConfiguration dlogConf = Utils.getDlogConf(workerConfig);
         try {
             this.dlogNamespace = NamespaceBuilder.newBuilder()
                     .conf(dlogConf)
                     .clientId("function-worker-" + workerConfig.getWorkerId())
-                    .uri(dlogUri)
+                    .uri(this.dlogUri)
                     .build();
         } catch (Exception e) {
             log.error("Failed to initialize dlog namespace {} for storing function packages",
@@ -128,7 +134,8 @@ public void start(URI dlogUri) throws InterruptedException {
             log.info("Created Pulsar client");
 
             //create scheduler manager
-            this.schedulerManager = new SchedulerManager(this.workerConfig, this.client);
+            this.schedulerManager = new SchedulerManager(this.workerConfig, this.client, this.brokerAdmin,
+                    this.executor);
 
             //create function meta data manager
             this.functionMetaDataManager = new FunctionMetaDataManager(
@@ -232,6 +239,10 @@ public void stop() {
         if (null != this.functionAdmin) {
             this.functionAdmin.close();
         }
+        
+        if(this.executor != null) {
+            this.executor.shutdown();
+        }
     }
 
 }
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/scheduler/RoundRobinScheduler.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/scheduler/RoundRobinScheduler.java
index 58c1a9a513..817962d4f1 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/scheduler/RoundRobinScheduler.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/scheduler/RoundRobinScheduler.java
@@ -22,6 +22,8 @@
 import org.apache.pulsar.functions.proto.Function.FunctionDetails;
 import org.apache.pulsar.functions.proto.Function.Instance;
 
+import com.google.common.collect.Lists;
+
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
@@ -47,19 +49,17 @@
             workerIdToAssignment.get(existingAssignment.getWorkerId()).add(existingAssignment);
         }
 
+        List<Assignment> newAssignments = Lists.newArrayList();
         for (Instance unassignedFunctionInstance : unassignedFunctionInstances) {
             String heartBeatWorkerId = checkHeartBeatFunction(unassignedFunctionInstance);
             String workerId = heartBeatWorkerId != null ? heartBeatWorkerId : findNextWorker(workerIdToAssignment);
             Assignment newAssignment = Assignment.newBuilder().setInstance(unassignedFunctionInstance)
                     .setWorkerId(workerId).build();
             workerIdToAssignment.get(workerId).add(newAssignment);
+            newAssignments.add(newAssignment);
         }
 
-        List<Assignment> assignments
-                = workerIdToAssignment.entrySet().stream()
-                .flatMap(entry -> entry.getValue().stream()).collect(Collectors.toList());
-
-        return assignments;
+        return newAssignments;
     }
 
     private static String checkHeartBeatFunction(Instance funInstance) {
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 85a2122dd0..5e1ed023aa 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
@@ -37,6 +37,7 @@
 import java.util.Map;
 
 import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyBoolean;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Matchers.argThat;
 import static org.mockito.Mockito.doReturn;
@@ -86,6 +87,8 @@ public void testProcessAssignmentUpdateAddFunctions() throws Exception {
         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(mock(Reader.class)).when(readerBuilder).create();
         WorkerService workerService = mock(WorkerService.class);
         doReturn(pulsarClient).when(workerService).getClient();
@@ -123,12 +126,8 @@ public void testProcessAssignmentUpdateAddFunctions() throws Exception {
         assignments.add(assignment1);
         assignments.add(assignment2);
 
-        Request.AssignmentsUpdate assignmentsUpdate = Request.AssignmentsUpdate.newBuilder()
-                .addAllAssignments(assignments)
-                .setVersion(1)
-                .build();
-
-        functionRuntimeManager.processAssignmentUpdate(MessageId.earliest, assignmentsUpdate);
+        functionRuntimeManager.processAssignment(assignment1);
+        functionRuntimeManager.processAssignment(assignment2);
 
         verify(functionRuntimeManager, times(2)).setAssignment(any(Function.Assignment.class));
         verify(functionRuntimeManager, times(0)).deleteAssignment(any(Function.Assignment.class));
@@ -183,6 +182,7 @@ public void testProcessAssignmentUpdateDeleteFunctions() throws Exception {
         doReturn(readerBuilder).when(pulsarClient).newReader();
         doReturn(readerBuilder).when(readerBuilder).topic(anyString());
         doReturn(readerBuilder).when(readerBuilder).startMessageId(any());
+        doReturn(readerBuilder).when(readerBuilder).readCompacted(anyBoolean());
         doReturn(mock(Reader.class)).when(readerBuilder).create();
         WorkerService workerService = mock(WorkerService.class);
         doReturn(pulsarClient).when(workerService).getClient();
@@ -205,6 +205,7 @@ public void testProcessAssignmentUpdateDeleteFunctions() throws Exception {
                 Function.FunctionDetails.newBuilder()
                         .setTenant("test-tenant").setNamespace("test-namespace").setName("func-2")).build();
 
+        // Delete this assignment
         Function.Assignment assignment1 = Function.Assignment.newBuilder()
                 .setWorkerId("worker-1")
                 .setInstance(Function.Instance.newBuilder()
@@ -221,23 +222,18 @@ public void testProcessAssignmentUpdateDeleteFunctions() throws Exception {
         functionRuntimeManager.setAssignment(assignment2);
         reset(functionRuntimeManager);
 
-        List<Function.Assignment> assignments = new LinkedList<>();
-        assignments.add(assignment2);
-
-        Request.AssignmentsUpdate assignmentsUpdate = Request.AssignmentsUpdate.newBuilder()
-                .addAllAssignments(assignments)
-                .setVersion(1)
-                .build();
-
         functionRuntimeManager.functionRuntimeInfoMap.put(
                 "test-tenant/test-namespace/func-1:0", new FunctionRuntimeInfo().setFunctionInstance(
                         Function.Instance.newBuilder().setFunctionMetaData(function1).setInstanceId(0)
                                 .build()));
 
-        functionRuntimeManager.processAssignmentUpdate(MessageId.earliest, assignmentsUpdate);
+        functionRuntimeManager.processAssignment(assignment1);
+        functionRuntimeManager.processAssignment(assignment2);
 
+        functionRuntimeManager.deleteAssignment(Utils.getFullyQualifiedInstanceId(assignment1.getInstance()));
+        
         verify(functionRuntimeManager, times(0)).setAssignment(any(Function.Assignment.class));
-        verify(functionRuntimeManager, times(1)).deleteAssignment(any(Function.Assignment.class));
+        verify(functionRuntimeManager, times(1)).deleteAssignment(any(String.class));
 
         Assert.assertEquals(functionRuntimeManager.workerIdToAssignments.size(), 1);
         Assert.assertEquals(functionRuntimeManager.workerIdToAssignments
@@ -284,6 +280,7 @@ public void testProcessAssignmentUpdateModifyFunctions() throws Exception {
         doReturn(readerBuilder).when(pulsarClient).newReader();
         doReturn(readerBuilder).when(readerBuilder).topic(anyString());
         doReturn(readerBuilder).when(readerBuilder).startMessageId(any());
+        doReturn(readerBuilder).when(readerBuilder).readCompacted(anyBoolean());
         doReturn(mock(Reader.class)).when(readerBuilder).create();
         WorkerService workerService = mock(WorkerService.class);
         doReturn(pulsarClient).when(workerService).getClient();
@@ -327,14 +324,6 @@ public void testProcessAssignmentUpdateModifyFunctions() throws Exception {
                 .setInstance(Function.Instance.newBuilder()
                         .setFunctionMetaData(function2).setInstanceId(0).build())
                 .build();
-        List<Function.Assignment> assignments = new LinkedList<>();
-        assignments.add(assignment1);
-        assignments.add(assignment3);
-
-        Request.AssignmentsUpdate assignmentsUpdate = Request.AssignmentsUpdate.newBuilder()
-                .addAllAssignments(assignments)
-                .setVersion(1)
-                .build();
 
         functionRuntimeManager.functionRuntimeInfoMap.put(
                 "test-tenant/test-namespace/func-1:0", new FunctionRuntimeInfo().setFunctionInstance(
@@ -345,7 +334,8 @@ public void testProcessAssignmentUpdateModifyFunctions() throws Exception {
                         Function.Instance.newBuilder().setFunctionMetaData(function2).setInstanceId(0)
                                 .build()));
 
-        functionRuntimeManager.processAssignmentUpdate(MessageId.earliest, assignmentsUpdate);
+        functionRuntimeManager.processAssignment(assignment1);
+        functionRuntimeManager.processAssignment(assignment3);
 
         verify(functionRuntimeManager, times(1)).insertStopAction(any(FunctionRuntimeInfo.class));
         verify(functionRuntimeManager).insertStopAction(argThat(new ArgumentMatcher<FunctionRuntimeInfo>() {
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/MembershipManagerTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/MembershipManagerTest.java
index 2753bf196a..c849fd36de 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/MembershipManagerTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/MembershipManagerTest.java
@@ -131,6 +131,7 @@ public void testCheckFailuresNoFailures() throws Exception {
         ReaderBuilder<byte[]> readerBuilder = mock(ReaderBuilder.class);
         doReturn(readerBuilder).when(pulsarClient).newReader();
         doReturn(readerBuilder).when(readerBuilder).topic(anyString());
+        doReturn(readerBuilder).when(readerBuilder).readCompacted(true);
         doReturn(readerBuilder).when(readerBuilder).startMessageId(any());
         doReturn(mock(Reader.class)).when(readerBuilder).create();
         WorkerService workerService = mock(WorkerService.class);
@@ -202,6 +203,7 @@ public void testCheckFailuresSomeFailures() throws Exception {
         doReturn(readerBuilder).when(pulsarClient).newReader();
         doReturn(readerBuilder).when(readerBuilder).topic(anyString());
         doReturn(readerBuilder).when(readerBuilder).startMessageId(any());
+        doReturn(readerBuilder).when(readerBuilder).readCompacted(true);
         doReturn(mock(Reader.class)).when(readerBuilder).create();
         WorkerService workerService = mock(WorkerService.class);
         doReturn(pulsarClient).when(workerService).getClient();
@@ -296,6 +298,7 @@ public void testCheckFailuresSomeUnassigned() throws Exception {
         ReaderBuilder<byte[]> readerBuilder = mock(ReaderBuilder.class);
         doReturn(readerBuilder).when(pulsarClient).newReader();
         doReturn(readerBuilder).when(readerBuilder).topic(anyString());
+        doReturn(readerBuilder).when(readerBuilder).readCompacted(true);
         doReturn(readerBuilder).when(readerBuilder).startMessageId(any());
         doReturn(mock(Reader.class)).when(readerBuilder).create();
         WorkerService workerService = mock(WorkerService.class);
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/SchedulerManagerTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/SchedulerManagerTest.java
index 19977bd812..00e4b6e0e1 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/SchedulerManagerTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/SchedulerManagerTest.java
@@ -28,15 +28,19 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertTrue;
 
 import java.lang.reflect.Method;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
@@ -46,6 +50,7 @@
 import org.apache.pulsar.client.api.ProducerBuilder;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.TypedMessageBuilder;
 import org.apache.pulsar.functions.proto.Function;
 import org.apache.pulsar.functions.proto.Function.Assignment;
 import org.apache.pulsar.functions.proto.Request;
@@ -53,9 +58,14 @@
 import org.mockito.Mockito;
 import org.mockito.invocation.Invocation;
 import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
+import com.google.common.collect.Sets;
+import com.google.protobuf.InvalidProtocolBufferException;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
@@ -67,6 +77,8 @@
     private MembershipManager membershipManager;
     private CompletableFuture<MessageId> completableFuture;
     private Producer producer;
+    private TypedMessageBuilder<byte[]> message;
+    private ScheduledExecutorService executor;
 
     private static PulsarClient mockPulsarClient() throws PulsarClientException {
         ProducerBuilder<byte[]> builder = mock(ProducerBuilder.class);
@@ -94,8 +106,12 @@ public void setup() throws PulsarClientException {
         producer = mock(Producer.class);
         completableFuture = spy(new CompletableFuture<>());
         completableFuture.complete(MessageId.earliest);
-        byte[] bytes = any();
-        when(producer.sendAsync(bytes)).thenReturn(completableFuture);
+        //byte[] bytes = any();
+        message = mock(TypedMessageBuilder.class);
+        when(producer.newMessage()).thenReturn(message);
+        when(message.key(anyString())).thenReturn(message);
+        when(message.value(any())).thenReturn(message);
+        when(message.sendAsync()).thenReturn(completableFuture);
 
         ProducerBuilder<byte[]> builder = mock(ProducerBuilder.class);
         when(builder.topic(anyString())).thenReturn(builder);
@@ -109,7 +125,9 @@ public void setup() throws PulsarClientException {
         PulsarClient pulsarClient = mock(PulsarClient.class);
         when(pulsarClient.newProducer()).thenReturn(builder);
 
-        schedulerManager = spy(new SchedulerManager(workerConfig, pulsarClient));
+        this.executor = Executors
+                .newSingleThreadScheduledExecutor(new DefaultThreadFactory("worker-test"));
+        schedulerManager = spy(new SchedulerManager(workerConfig, pulsarClient, null, executor));
         functionRuntimeManager = mock(FunctionRuntimeManager.class);
         functionMetaDataManager = mock(FunctionMetaDataManager.class);
         membershipManager = mock(MembershipManager.class);
@@ -118,6 +136,11 @@ public void setup() throws PulsarClientException {
         schedulerManager.setMembershipManager(membershipManager);
     }
 
+    @AfterMethod
+    public void stop() {
+        this.executor.shutdown();
+    }
+    
     @Test
     public void testSchedule() throws Exception {
 
@@ -143,9 +166,6 @@ public void testSchedule() throws Exception {
         currentAssignments.put("worker-1", assignmentEntry1);
         doReturn(currentAssignments).when(functionRuntimeManager).getCurrentAssignments();
 
-        //set version
-        doReturn(version).when(functionRuntimeManager).getCurrentAssignmentVersion();
-
         // single node
         List<WorkerInfo> workerInfoList = new LinkedList<>();
         workerInfoList.add(WorkerInfo.of("worker-1", "workerHostname-1", 5000));
@@ -159,7 +179,9 @@ public void testSchedule() throws Exception {
         // i am leader
         doReturn(true).when(membershipManager).isLeader();
         callSchedule();
-        verify(producer, times(1)).sendAsync(any(byte[].class));
+        List<Invocation> invocations = getMethodInvocationDetails(schedulerManager,
+                SchedulerManager.class.getMethod("invokeScheduler"));
+        Assert.assertEquals(invocations.size(), 1);
     }
 
     @Test
@@ -187,9 +209,6 @@ public void testNothingNewToSchedule() throws Exception {
         currentAssignments.put("worker-1", assignmentEntry1);
         doReturn(currentAssignments).when(functionRuntimeManager).getCurrentAssignments();
 
-        //set version
-        doReturn(version).when(functionRuntimeManager).getCurrentAssignmentVersion();
-
         // single node
         List<WorkerInfo> workerInfoList = new LinkedList<>();
         workerInfoList.add(WorkerInfo.of("worker-1", "workerHostname-1", 5000));
@@ -200,17 +219,8 @@ public void testNothingNewToSchedule() throws Exception {
 
         callSchedule();
 
-        List<Invocation> invocations = getMethodInvocationDetails(producer, Producer.class.getMethod("sendAsync",
-                Object.class));
-        Assert.assertEquals(invocations.size(), 1);
-
-        byte[] send = (byte[]) invocations.get(0).getRawArguments()[0];
-        Request.AssignmentsUpdate assignmentsUpdate = Request.AssignmentsUpdate.parseFrom(send);
-        log.info("assignmentsUpdate: {}", assignmentsUpdate);
-        Assert.assertEquals(
-                Request.AssignmentsUpdate.newBuilder().setVersion(version + 1)
-                        .addAssignments(assignment1).build(),
-                assignmentsUpdate);
+        List<Invocation> invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("sendAsync"));
+        Assert.assertEquals(invocations.size(), 0);
     }
 
     @Test
@@ -243,9 +253,6 @@ public void testAddingFunctions() throws Exception {
         currentAssignments.put("worker-1", assignmentEntry1);
         doReturn(currentAssignments).when(functionRuntimeManager).getCurrentAssignments();
 
-        //set version
-        doReturn(version).when(functionRuntimeManager).getCurrentAssignmentVersion();
-
         // single node
         List<WorkerInfo> workerInfoList = new LinkedList<>();
         workerInfoList.add(WorkerInfo.of("worker-1", "workerHostname-1", 5000));
@@ -256,23 +263,20 @@ public void testAddingFunctions() throws Exception {
 
         callSchedule();
 
-        List<Invocation> invocations = getMethodInvocationDetails(producer, Producer.class.getMethod("sendAsync",
-                Object.class));
+        List<Invocation> invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("sendAsync"));
         Assert.assertEquals(invocations.size(), 1);
-
+        invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("value",
+                Object.class));
         byte[] send = (byte[]) invocations.get(0).getRawArguments()[0];
-        Request.AssignmentsUpdate assignmentsUpdate = Request.AssignmentsUpdate.parseFrom(send);
+        Assignment assignments = Assignment.parseFrom(send);
 
-        log.info("assignmentsUpdate: {}", assignmentsUpdate);
+        log.info("assignments: {}", assignments);
         Function.Assignment assignment2 = Function.Assignment.newBuilder()
                 .setWorkerId("worker-1")
                 .setInstance(Function.Instance.newBuilder()
                         .setFunctionMetaData(function2).setInstanceId(0).build())
                 .build();
-        Assert.assertEquals(
-                Request.AssignmentsUpdate.newBuilder().setVersion(version + 1)
-                        .addAssignments(assignment1).addAssignments(assignment2).build(),
-                assignmentsUpdate);
+        Assert.assertEquals(assignment2, assignments);
 
     }
 
@@ -288,7 +292,7 @@ public void testDeletingFunctions() throws Exception {
         // simulate function2 got removed
         Function.FunctionMetaData function2 = Function.FunctionMetaData.newBuilder()
                 .setFunctionDetails(Function.FunctionDetails.newBuilder().setName("func-2")
-                        .setNamespace("namespace-1").setTenant("tenant-1").setParallelism(1)).setVersion(version)
+                        .setNamespace("namespace-1").setTenant("tenant-1").setParallelism(1))
                 .build();
         functionMetaDataList.add(function1);
         doReturn(functionMetaDataList).when(functionMetaDataManager).getAllFunctionMetaData();
@@ -300,6 +304,7 @@ public void testDeletingFunctions() throws Exception {
                         .setFunctionMetaData(function1).setInstanceId(0).build())
                 .build();
 
+        // Delete this assignment
         Function.Assignment assignment2 = Function.Assignment.newBuilder()
                 .setWorkerId("worker-1")
                 .setInstance(Function.Instance.newBuilder()
@@ -309,14 +314,12 @@ public void testDeletingFunctions() throws Exception {
         Map<String, Map<String, Function.Assignment>> currentAssignments = new HashMap<>();
         Map<String, Function.Assignment> assignmentEntry1 = new HashMap<>();
         assignmentEntry1.put(Utils.getFullyQualifiedInstanceId(assignment1.getInstance()), assignment1);
+        //TODO: delete this assignment
         assignmentEntry1.put(Utils.getFullyQualifiedInstanceId(assignment2.getInstance()), assignment2);
 
         currentAssignments.put("worker-1", assignmentEntry1);
         doReturn(currentAssignments).when(functionRuntimeManager).getCurrentAssignments();
 
-        //set version
-        doReturn(version).when(functionRuntimeManager).getCurrentAssignmentVersion();
-
         // single node
         List<WorkerInfo> workerInfoList = new LinkedList<>();
         workerInfoList.add(WorkerInfo.of("worker-1", "workerHostname-1", 5000));
@@ -327,19 +330,16 @@ public void testDeletingFunctions() throws Exception {
 
         callSchedule();
 
-        List<Invocation> invocations = getMethodInvocationDetails(producer, Producer.class.getMethod("sendAsync",
-                Object.class));
+        List<Invocation> invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("sendAsync"));
         Assert.assertEquals(invocations.size(), 1);
-
+        invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("value",
+                Object.class));
         byte[] send = (byte[]) invocations.get(0).getRawArguments()[0];
-        Request.AssignmentsUpdate assignmentsUpdate = Request.AssignmentsUpdate.parseFrom(send);
+        Assignment assignments = Assignment.parseFrom(send);
 
-        log.info("assignmentsUpdate: {}", assignmentsUpdate);
+        log.info("assignments: {}", assignments);
 
-        Assert.assertEquals(
-                Request.AssignmentsUpdate.newBuilder().setVersion(version + 1)
-                        .addAssignments(assignment1).build(),
-                assignmentsUpdate);
+        Assert.assertEquals(0, send.length);
     }
 
     @Test
@@ -373,9 +373,6 @@ public void testScalingUp() throws Exception {
         currentAssignments.put("worker-1", assignmentEntry1);
         doReturn(currentAssignments).when(functionRuntimeManager).getCurrentAssignments();
 
-        //set version
-        doReturn(version).when(functionRuntimeManager).getCurrentAssignmentVersion();
-
         // single node
         List<WorkerInfo> workerInfoList = new LinkedList<>();
         workerInfoList.add(WorkerInfo.of("worker-1", "workerHostname-1", 5000));
@@ -386,25 +383,21 @@ public void testScalingUp() throws Exception {
 
         callSchedule();
 
-        List<Invocation> invocations = getMethodInvocationDetails(producer, Producer.class.getMethod("sendAsync",
-                Object.class));
+        List<Invocation> invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("sendAsync"));
         Assert.assertEquals(invocations.size(), 1);
-
+        invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("value",
+                Object.class));
         byte[] send = (byte[]) invocations.get(0).getRawArguments()[0];
-        Request.AssignmentsUpdate assignmentsUpdate = Request.AssignmentsUpdate.parseFrom(send);
+        Assignment assignments = Assignment.parseFrom(send);
 
-        log.info("assignmentsUpdate: {}", assignmentsUpdate);
+        log.info("assignments: {}", assignments);
 
         Function.Assignment assignment2 = Function.Assignment.newBuilder()
                 .setWorkerId("worker-1")
                 .setInstance(Function.Instance.newBuilder()
                         .setFunctionMetaData(function2).setInstanceId(0).build())
                 .build();
-        Assert.assertEquals(
-                assignmentsUpdate,
-                Request.AssignmentsUpdate.newBuilder().setVersion(version + 1)
-                        .addAssignments(assignment1).addAssignments(assignment2).build()
-                );
+        Assert.assertEquals(assignments, assignment2);
 
         // scale up
 
@@ -435,17 +428,23 @@ public void testScalingUp() throws Exception {
 
         callSchedule();
 
-        invocations = getMethodInvocationDetails(producer, Producer.class.getMethod("sendAsync",
+        invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("sendAsync"));
+        Assert.assertEquals(invocations.size(), 4);
+        invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("value",
                 Object.class));
-        Assert.assertEquals(invocations.size(), 2);
+        
+        Set<Assignment> allAssignments = Sets.newHashSet();
+        invocations.forEach(invocation -> {
+            try {
+                allAssignments.add(Assignment.parseFrom((byte[])invocation.getRawArguments()[0]));
+            } catch (InvalidProtocolBufferException e) {
+                throw new RuntimeException(e);
+            }
+        });
 
-        send = (byte[]) invocations.get(1).getRawArguments()[0];
-        assignmentsUpdate = Request.AssignmentsUpdate.parseFrom(send);
-        Assert.assertEquals(assignmentsUpdate,
-                Request.AssignmentsUpdate.newBuilder().setVersion(version + 1 + 1)
-                        .addAssignments(assignment1).addAssignments(assignment2Scaled1)
-                        .addAssignments(assignment2Scaled2).addAssignments(assignment2Scaled3).build()
-                );
+        assertTrue(allAssignments.contains(assignment2Scaled1));
+        assertTrue(allAssignments.contains(assignment2Scaled2));
+        assertTrue(allAssignments.contains(assignment2Scaled3));
     }
 
     @Test
@@ -479,9 +478,6 @@ public void testScalingDown() throws Exception {
         currentAssignments.put("worker-1", assignmentEntry1);
         doReturn(currentAssignments).when(functionRuntimeManager).getCurrentAssignments();
 
-        //set version
-        doReturn(version).when(functionRuntimeManager).getCurrentAssignmentVersion();
-
         // single node
         List<WorkerInfo> workerInfoList = new LinkedList<>();
         workerInfoList.add(WorkerInfo.of("worker-1", "workerHostname-1", 5000));
@@ -492,14 +488,24 @@ public void testScalingDown() throws Exception {
 
         callSchedule();
 
-        List<Invocation> invocations = getMethodInvocationDetails(producer, Producer.class.getMethod("sendAsync",
+        List<Invocation> invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("sendAsync"));
+        Assert.assertEquals(invocations.size(), 3);
+        invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("value",
                 Object.class));
-        Assert.assertEquals(invocations.size(), 1);
-
         byte[] send = (byte[]) invocations.get(0).getRawArguments()[0];
-        Request.AssignmentsUpdate assignmentsUpdate = Request.AssignmentsUpdate.parseFrom(send);
+        Assignment assignments = Assignment.parseFrom(send);
+
+        log.info("assignments: {}", assignments);
+        
+        Set<Assignment> allAssignments = Sets.newHashSet();
+        invocations.forEach(invocation -> {
+            try {
+                allAssignments.add(Assignment.parseFrom((byte[])invocation.getRawArguments()[0]));
+            } catch (InvalidProtocolBufferException e) {
+                throw new RuntimeException(e);
+            }
+        });
 
-        log.info("assignmentsUpdate: {}", assignmentsUpdate);
 
         Function.Assignment assignment2_1 = Function.Assignment.newBuilder()
                 .setWorkerId("worker-1")
@@ -516,13 +522,11 @@ public void testScalingDown() throws Exception {
                 .setInstance(Function.Instance.newBuilder()
                         .setFunctionMetaData(function2).setInstanceId(2).build())
                 .build();
-        Assert.assertEquals(
-                assignmentsUpdate,
-                Request.AssignmentsUpdate.newBuilder().setVersion(version + 1)
-                        .addAssignments(assignment1).addAssignments(assignment2_1)
-                        .addAssignments(assignment2_2).addAssignments(assignment2_3).build()
-        );
-
+        
+        assertTrue(allAssignments.contains(assignment2_1));
+        assertTrue(allAssignments.contains(assignment2_2));
+        assertTrue(allAssignments.contains(assignment2_3));
+        
         // scale down
 
         Function.FunctionMetaData function2Scaled = Function.FunctionMetaData.newBuilder()
@@ -542,17 +546,23 @@ public void testScalingDown() throws Exception {
 
         callSchedule();
 
-        invocations = getMethodInvocationDetails(producer, Producer.class.getMethod("sendAsync",
+        invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("sendAsync"));
+        Assert.assertEquals(invocations.size(), 4);
+        invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("value",
                 Object.class));
-        Assert.assertEquals(invocations.size(), 2);
+        send = (byte[]) invocations.get(0).getRawArguments()[0];
+        assignments = Assignment.parseFrom(send);
+        
+        Set<Assignment> allAssignments2 = Sets.newHashSet();
+        invocations.forEach(invocation -> {
+            try {
+                allAssignments2.add(Assignment.parseFrom((byte[])invocation.getRawArguments()[0]));
+            } catch (InvalidProtocolBufferException e) {
+                throw new RuntimeException(e);
+            }
+        });
 
-        send = (byte[]) invocations.get(1).getRawArguments()[0];
-        assignmentsUpdate = Request.AssignmentsUpdate.parseFrom(send);
-        Assert.assertEquals(assignmentsUpdate,
-                Request.AssignmentsUpdate.newBuilder().setVersion(version + 1 + 1)
-                        .addAssignments(assignment1).addAssignments(assignment2Scaled)
-                        .build()
-        );
+        assertTrue(allAssignments2.contains(assignment2Scaled));
     }
 
     @Test
@@ -582,9 +592,6 @@ public void testHeartbeatFunction() throws Exception {
         currentAssignments.put("worker-1", assignmentEntry1);
         doReturn(currentAssignments).when(functionRuntimeManager).getCurrentAssignments();
 
-        // set version
-        doReturn(version).when(functionRuntimeManager).getCurrentAssignmentVersion();
-
         List<WorkerInfo> workerInfoList = new LinkedList<>();
         workerInfoList.add(WorkerInfo.of(workerId1, "workerHostname-1", 5000));
         workerInfoList.add(WorkerInfo.of(workerId2, "workerHostname-1", 6000));
@@ -595,20 +602,20 @@ public void testHeartbeatFunction() throws Exception {
 
         callSchedule();
 
-        List<Invocation> invocations = getMethodInvocationDetails(producer,
-                Producer.class.getMethod("sendAsync", Object.class));
-        Assert.assertEquals(invocations.size(), 1);
-
-        byte[] send = (byte[]) invocations.get(0).getRawArguments()[0];
-        Request.AssignmentsUpdate assignmentsUpdate = Request.AssignmentsUpdate.parseFrom(send);
-
-        List<Assignment> assignmentList = assignmentsUpdate.getAssignmentsList();
-        Assert.assertEquals(assignmentList.size(), 2);
-        for (Assignment assignment : assignmentList) {
-            String functionName = assignment.getInstance().getFunctionMetaData().getFunctionDetails().getName();
-            String assignedWorkerId = assignment.getWorkerId();
-            Assert.assertEquals(functionName, assignedWorkerId);
-        }
+        List<Invocation> invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("sendAsync"));
+        Assert.assertEquals(invocations.size(), 2);
+        invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("value",
+                Object.class));
+        invocations.forEach(invocation -> {
+            try {
+                Assignment assignment = Assignment.parseFrom((byte[])invocation.getRawArguments()[0]);
+                String functionName = assignment.getInstance().getFunctionMetaData().getFunctionDetails().getName();
+                String assignedWorkerId = assignment.getWorkerId();
+                Assert.assertEquals(functionName, assignedWorkerId);
+            } catch (InvalidProtocolBufferException e) {
+                throw new RuntimeException(e);
+            }
+        });
     }
     
     @Test
@@ -644,9 +651,6 @@ public void testUpdate() throws Exception {
         currentAssignments.put("worker-1", assignmentEntry1);
         doReturn(currentAssignments).when(functionRuntimeManager).getCurrentAssignments();
 
-        //set version
-        doReturn(version).when(functionRuntimeManager).getCurrentAssignmentVersion();
-
         // single node
         List<WorkerInfo> workerInfoList = new LinkedList<>();
         workerInfoList.add(WorkerInfo.of("worker-1", "workerHostname-1", 5000));
@@ -657,14 +661,14 @@ public void testUpdate() throws Exception {
 
         callSchedule();
 
-        List<Invocation> invocations = getMethodInvocationDetails(producer, Producer.class.getMethod("sendAsync",
+        List<Invocation> invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("sendAsync"));
+        Assert.assertEquals(invocations.size(), 3);
+        invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("value",
                 Object.class));
-        Assert.assertEquals(invocations.size(), 1);
-
         byte[] send = (byte[]) invocations.get(0).getRawArguments()[0];
-        Request.AssignmentsUpdate assignmentsUpdate = Request.AssignmentsUpdate.parseFrom(send);
+        Assignment assignments = Assignment.parseFrom(send);
 
-        log.info("assignmentsUpdate: {}", assignmentsUpdate);
+        log.info("assignmentsUpdate: {}", assignments);
 
         Function.Assignment assignment2_1 = Function.Assignment.newBuilder()
                 .setWorkerId("worker-1")
@@ -681,13 +685,27 @@ public void testUpdate() throws Exception {
                 .setInstance(Function.Instance.newBuilder()
                         .setFunctionMetaData(function2).setInstanceId(2).build())
                 .build();
-        Assert.assertEquals(
-                assignmentsUpdate,
-                Request.AssignmentsUpdate.newBuilder().setVersion(version + 1)
-                        .addAssignments(assignment1).addAssignments(assignment2_1)
-                        .addAssignments(assignment2_2).addAssignments(assignment2_3).build()
-        );
-
+        
+        invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("sendAsync"));
+        Assert.assertEquals(invocations.size(), 3);
+        invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("value",
+                Object.class));
+        send = (byte[]) invocations.get(0).getRawArguments()[0];
+        assignments = Assignment.parseFrom(send);
+        
+        Set<Assignment> allAssignments = Sets.newHashSet();
+        invocations.forEach(invocation -> {
+            try {
+                allAssignments.add(Assignment.parseFrom((byte[])invocation.getRawArguments()[0]));
+            } catch (InvalidProtocolBufferException e) {
+                throw new RuntimeException(e);
+            }
+        });
+        
+        assertTrue(allAssignments.contains(assignment2_1));
+        assertTrue(allAssignments.contains(assignment2_2));
+        assertTrue(allAssignments.contains(assignment2_3));
+        
         // scale down
 
         Function.FunctionMetaData function2Updated = Function.FunctionMetaData.newBuilder()
@@ -718,28 +736,32 @@ public void testUpdate() throws Exception {
 
         callSchedule();
 
-        invocations = getMethodInvocationDetails(producer, Producer.class.getMethod("sendAsync",
+        invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("sendAsync"));
+        Assert.assertEquals(invocations.size(), 6);
+        invocations = getMethodInvocationDetails(message, TypedMessageBuilder.class.getMethod("value",
                 Object.class));
-        Assert.assertEquals(invocations.size(), 2);
-
-        send = (byte[]) invocations.get(1).getRawArguments()[0];
-        assignmentsUpdate = Request.AssignmentsUpdate.parseFrom(send);
-        Assert.assertEquals(assignmentsUpdate,
-                Request.AssignmentsUpdate.newBuilder().setVersion(version + 1 + 1)
-                        .addAssignments(assignment1).addAssignments(assignment2Updated1)
-                        .addAssignments(assignment2Updated2)
-                        .addAssignments(assignment2Updated3)
-                        .build()
-        );
+        send = (byte[]) invocations.get(0).getRawArguments()[0];
+        assignments = Assignment.parseFrom(send);
+        
+        Set<Assignment> allAssignments2 = Sets.newHashSet();
+        invocations.forEach(invocation -> {
+            try {
+                allAssignments2.add(Assignment.parseFrom((byte[])invocation.getRawArguments()[0]));
+            } catch (InvalidProtocolBufferException e) {
+                throw new RuntimeException(e);
+            }
+        });
+        
+        assertTrue(allAssignments2.contains(assignment2Updated1));
+        assertTrue(allAssignments2.contains(assignment2Updated2));
+        assertTrue(allAssignments2.contains(assignment2Updated3));
     }
 
     private void callSchedule() throws NoSuchMethodException, InterruptedException,
             TimeoutException, ExecutionException {
-        long intialVersion = functionRuntimeManager.getCurrentAssignmentVersion();
         Future<?> complete = schedulerManager.schedule();
 
         complete.get(30, TimeUnit.SECONDS);
-        doReturn(intialVersion + 1).when(functionRuntimeManager).getCurrentAssignmentVersion();
     }
 
     private List<Invocation> getMethodInvocationDetails(Object o, Method method) throws NoSuchMethodException {


 

----------------------------------------------------------------
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:
users@infra.apache.org


With regards,
Apache Git Services

Mime
View raw message