@Test public void processUpdateTest() throws PulsarClientException { SchedulerManager schedulerManager = mock(SchedulerManager.class); WorkerConfig workerConfig = new WorkerConfig(); workerConfig.setWorkerId("worker-1"); FunctionMetaDataManager functionMetaDataManager = spy( new FunctionMetaDataManager(workerConfig, schedulerManager, mockPulsarClient(), ErrorNotifier.getDefaultImpl())); Function.FunctionMetaData m1 = Function.FunctionMetaData.newBuilder() .setVersion(1) .setFunctionDetails(Function.FunctionDetails.newBuilder().setName("func-1") .setNamespace("namespace-1").setTenant("tenant-1")).build(); Assert.assertTrue(functionMetaDataManager.processUpdate(m1)); verify(functionMetaDataManager, times(1)) .setFunctionMetaData(any(Function.FunctionMetaData.class)); verify(schedulerManager, times(0)).schedule(); Assert.assertEquals(m1, functionMetaDataManager.functionMetaDataMap.get( "tenant-1").get("namespace-1").get("func-1")); Assert.assertEquals(1, functionMetaDataManager.functionMetaDataMap.get( "tenant-1").get("namespace-1").size()); try { functionMetaDataManager.processUpdate(m1); Assert.assertTrue(false); } catch (IllegalArgumentException e) { Assert.assertEquals(e.getMessage(), "Update request ignored because it is out of date. Please try again."); } verify(functionMetaDataManager, times(1)) .setFunctionMetaData(any(Function.FunctionMetaData.class)); verify(schedulerManager, times(0)).schedule(); Assert.assertEquals(m1, functionMetaDataManager.functionMetaDataMap.get( "tenant-1").get("namespace-1").get("func-1")); Assert.assertEquals(1, functionMetaDataManager.functionMetaDataMap.get( "tenant-1").get("namespace-1").size()); m1 = m1.toBuilder().setVersion(2).build(); Assert.assertTrue(functionMetaDataManager.processUpdate(m1)); verify(functionMetaDataManager, times(2)) .setFunctionMetaData(any(Function.FunctionMetaData.class)); verify(schedulerManager, times(0)).schedule(); Assert.assertEquals(m1, functionMetaDataManager.functionMetaDataMap.get( "tenant-1").get("namespace-1").get("func-1")); Assert.assertEquals(1, functionMetaDataManager.functionMetaDataMap.get( "tenant-1").get("namespace-1").size()); }
@VisibleForTesting synchronized boolean processUpdate(FunctionMetaData updateRequestFs) throws IllegalArgumentException { log.debug("Process update request: {}", updateRequestFs); boolean needsScheduling = false; if (!this.containsFunctionMetaData(updateRequestFs)) { setFunctionMetaData(updateRequestFs); needsScheduling = true; } else { if (!isRequestOutdated(updateRequestFs)) { setFunctionMetaData(updateRequestFs); needsScheduling = true; } else { throw new IllegalArgumentException("Update request ignored because it is out of date. Please try again."); } } return needsScheduling; }
FunctionMetaDataManager implements AutoCloseable { @VisibleForTesting synchronized boolean processUpdate(FunctionMetaData updateRequestFs) throws IllegalArgumentException { log.debug("Process update request: {}", updateRequestFs); boolean needsScheduling = false; if (!this.containsFunctionMetaData(updateRequestFs)) { setFunctionMetaData(updateRequestFs); needsScheduling = true; } else { if (!isRequestOutdated(updateRequestFs)) { setFunctionMetaData(updateRequestFs); needsScheduling = true; } else { throw new IllegalArgumentException("Update request ignored because it is out of date. Please try again."); } } return needsScheduling; } }
FunctionMetaDataManager implements AutoCloseable { @VisibleForTesting synchronized boolean processUpdate(FunctionMetaData updateRequestFs) throws IllegalArgumentException { log.debug("Process update request: {}", updateRequestFs); boolean needsScheduling = false; if (!this.containsFunctionMetaData(updateRequestFs)) { setFunctionMetaData(updateRequestFs); needsScheduling = true; } else { if (!isRequestOutdated(updateRequestFs)) { setFunctionMetaData(updateRequestFs); needsScheduling = true; } else { throw new IllegalArgumentException("Update request ignored because it is out of date. Please try again."); } } return needsScheduling; } FunctionMetaDataManager(WorkerConfig workerConfig,
                                   SchedulerManager schedulerManager,
                                   PulsarClient pulsarClient,
                                   ErrorNotifier errorNotifier); }
FunctionMetaDataManager implements AutoCloseable { @VisibleForTesting synchronized boolean processUpdate(FunctionMetaData updateRequestFs) throws IllegalArgumentException { log.debug("Process update request: {}", updateRequestFs); boolean needsScheduling = false; if (!this.containsFunctionMetaData(updateRequestFs)) { setFunctionMetaData(updateRequestFs); needsScheduling = true; } else { if (!isRequestOutdated(updateRequestFs)) { setFunctionMetaData(updateRequestFs); needsScheduling = true; } else { throw new IllegalArgumentException("Update request ignored because it is out of date. Please try again."); } } return needsScheduling; } FunctionMetaDataManager(WorkerConfig workerConfig,
                                   SchedulerManager schedulerManager,
                                   PulsarClient pulsarClient,
                                   ErrorNotifier errorNotifier); synchronized void initialize(); synchronized void start(); @Override void close(); synchronized FunctionMetaData getFunctionMetaData(String tenant, String namespace, String functionName); synchronized List<FunctionMetaData> getAllFunctionMetaData(); synchronized Collection<FunctionMetaData> listFunctions(String tenant, String namespace); synchronized boolean containsFunction(String tenant, String namespace, String functionName); synchronized void updateFunctionOnLeader(FunctionMetaData functionMetaData, boolean delete); void acquireLeadership(); synchronized void giveupLeadership(); void processMetaDataTopicMessage(Message<byte[]> message); }
FunctionMetaDataManager implements AutoCloseable { @VisibleForTesting synchronized boolean processUpdate(FunctionMetaData updateRequestFs) throws IllegalArgumentException { log.debug("Process update request: {}", updateRequestFs); boolean needsScheduling = false; if (!this.containsFunctionMetaData(updateRequestFs)) { setFunctionMetaData(updateRequestFs); needsScheduling = true; } else { if (!isRequestOutdated(updateRequestFs)) { setFunctionMetaData(updateRequestFs); needsScheduling = true; } else { throw new IllegalArgumentException("Update request ignored because it is out of date. Please try again."); } } return needsScheduling; } FunctionMetaDataManager(WorkerConfig workerConfig,
                                   SchedulerManager schedulerManager,
                                   PulsarClient pulsarClient,
                                   ErrorNotifier errorNotifier); synchronized void initialize(); synchronized void start(); @Override void close(); synchronized FunctionMetaData getFunctionMetaData(String tenant, String namespace, String functionName); synchronized List<FunctionMetaData> getAllFunctionMetaData(); synchronized Collection<FunctionMetaData> listFunctions(String tenant, String namespace); synchronized boolean containsFunction(String tenant, String namespace, String functionName); synchronized void updateFunctionOnLeader(FunctionMetaData functionMetaData, boolean delete); void acquireLeadership(); synchronized void giveupLeadership(); void processMetaDataTopicMessage(Message<byte[]> message);  }