fix bug in scheduler (#260)

pulsar-functions/conf/function_worker.yml

@@ -23,7 +23,6 @@ workerPort: 6750
 functionMetadataTopicName: metadata
 functionMetadataSnapshotsTopicPath: snapshots
 clusterCoordinationTopicName: coordinate
-snapshotFreqMs: 600000
 pulsarFunctionsNamespace: sample/standalone/functions
 pulsarServiceUrl: pulsar://localhost:6650
 pulsarWebServiceUrl: http://localhost:8080

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java

@@ -132,9 +132,6 @@ public class SchedulerManager implements AutoCloseable {
                 .flatMap(stringMapEntry -> stringMapEntry.getValue().values().stream()).collect(Collectors.toList());
 
         List<Function.Instance> needsAssignment = this.getUnassignedFunctionInstances(workerIdToAssignments, allInstances);
-        if (needsAssignment.isEmpty()) {
-            return;
-        }
 
         List<Assignment> assignments = this.scheduler.schedule(
                 needsAssignment, currentAssignments, currentMembership);