@Test(timeOut = 10000) public void testTriggerReadToTheEndAndExit() throws Exception { WorkerConfig workerConfig = new WorkerConfig(); workerConfig.setWorkerId("worker-1"); workerConfig.setFunctionRuntimeFactoryClassName(ThreadRuntimeFactory.class.getName()); workerConfig.setFunctionRuntimeFactoryConfigs( ObjectMapperFactory.getThreadLocal().convertValue( new ThreadRuntimeFactoryConfig().setThreadGroupName("test"), Map.class)); workerConfig.setPulsarServiceUrl("pulsar: workerConfig.setStateStorageServiceUrl("foo"); workerConfig.setFunctionAssignmentTopicName("assignments"); Function.FunctionMetaData function1 = Function.FunctionMetaData.newBuilder().setFunctionDetails( Function.FunctionDetails.newBuilder() .setTenant("test-tenant").setNamespace("test-namespace").setName("func-1")).build(); Function.FunctionMetaData function2 = Function.FunctionMetaData.newBuilder().setFunctionDetails( Function.FunctionDetails.newBuilder() .setTenant("test-tenant").setNamespace("test-namespace").setName("func-2")).build(); Function.Assignment assignment1 = Function.Assignment.newBuilder() .setWorkerId("worker-1") .setInstance(Function.Instance.newBuilder() .setFunctionMetaData(function1).setInstanceId(0).build()) .build(); Function.Assignment assignment2 = Function.Assignment.newBuilder() .setWorkerId("worker-1") .setInstance(Function.Instance.newBuilder() .setFunctionMetaData(function2).setInstanceId(0).build()) .build(); ArrayBlockingQueue<Message<byte[]>> messageList = new ArrayBlockingQueue<>(2); MessageId messageId1 = new MessageIdImpl(1, 1, -1); MessageId messageId2 = new MessageIdImpl(1, 2, -1); PulsarApi.MessageMetadata.Builder msgMetadataBuilder = PulsarApi.MessageMetadata.newBuilder(); Message message1 = spy(new MessageImpl("foo", messageId1.toString(), new HashMap<>(), Unpooled.copiedBuffer(assignment1.toByteArray()), null, msgMetadataBuilder)); doReturn(FunctionCommon.getFullyQualifiedInstanceId(assignment1.getInstance())).when(message1).getKey(); Message message2 = spy(new MessageImpl("foo", messageId2.toString(), new HashMap<>(), Unpooled.copiedBuffer(assignment2.toByteArray()), null, msgMetadataBuilder)); doReturn(FunctionCommon.getFullyQualifiedInstanceId(assignment2.getInstance())).when(message2).getKey(); PulsarClient pulsarClient = mock(PulsarClient.class); Reader<byte[]> reader = mock(Reader.class); when(reader.readNext(anyInt(), any())).thenAnswer(new Answer<Message<byte[]>>() { @Override public Message<byte[]> answer(InvocationOnMock invocationOnMock) throws Throwable { return messageList.poll(10, TimeUnit.MILLISECONDS); } }); when(reader.readNextAsync()).thenAnswer(new Answer<CompletableFuture<Message<byte[]>>>() { @Override public CompletableFuture<Message<byte[]>> answer(InvocationOnMock invocationOnMock) throws Throwable { return new CompletableFuture<>(); } }); when(reader.hasMessageAvailable()).thenAnswer(new Answer<Boolean>() { @Override public Boolean answer(InvocationOnMock invocationOnMock) throws Throwable { return !messageList.isEmpty(); } }); ReaderBuilder readerBuilder = mock(ReaderBuilder.class); doReturn(readerBuilder).when(pulsarClient).newReader(); doReturn(readerBuilder).when(readerBuilder).topic(anyString()); doReturn(readerBuilder).when(readerBuilder).readerName(anyString()); doReturn(readerBuilder).when(readerBuilder).subscriptionRolePrefix(anyString()); doReturn(readerBuilder).when(readerBuilder).startMessageId(any()); doReturn(readerBuilder).when(readerBuilder).startMessageId(any()); doReturn(readerBuilder).when(readerBuilder).readCompacted(anyBoolean()); doReturn(reader).when(readerBuilder).create(); WorkerService workerService = mock(WorkerService.class); doReturn(pulsarClient).when(workerService).getClient(); doReturn(mock(PulsarAdmin.class)).when(workerService).getFunctionAdmin(); ErrorNotifier errorNotifier = spy(ErrorNotifier.getDefaultImpl()); FunctionRuntimeManager functionRuntimeManager = mock(FunctionRuntimeManager.class); FunctionAssignmentTailer functionAssignmentTailer = spy(new FunctionAssignmentTailer(functionRuntimeManager, readerBuilder, workerConfig, errorNotifier)); functionAssignmentTailer.start(); messageList.add(message1); for (int i = 0; i < 10; i++) { try { verify(functionRuntimeManager, times(1)).processAssignmentMessage(eq(message1)); break; } catch (org.mockito.exceptions.verification.WantedButNotInvoked e) { if (i == 9) { throw e; } } Thread.sleep(200); } functionAssignmentTailer.triggerReadToTheEndAndExit().get(); for (int i = 0; i < 10; i++) { if(!functionAssignmentTailer.getThread().isAlive()) { break; } if (i == 9) { Assert.assertFalse(functionAssignmentTailer.getThread().isAlive()); } Thread.sleep(200); } messageList.add(message2); Assert.assertEquals(functionAssignmentTailer.getLastMessageId(), message1.getMessageId()); functionAssignmentTailer.close(); }
public synchronized CompletableFuture<Void> triggerReadToTheEndAndExit() { exitOnEndOfTopic = true; return this.exitFuture; }
FunctionAssignmentTailer implements AutoCloseable { public synchronized CompletableFuture<Void> triggerReadToTheEndAndExit() { exitOnEndOfTopic = true; return this.exitFuture; } }
FunctionAssignmentTailer implements AutoCloseable { public synchronized CompletableFuture<Void> triggerReadToTheEndAndExit() { exitOnEndOfTopic = true; return this.exitFuture; } FunctionAssignmentTailer(
            FunctionRuntimeManager functionRuntimeManager,
            ReaderBuilder readerBuilder,
            WorkerConfig workerConfig,
            ErrorNotifier errorNotifier); }
FunctionAssignmentTailer implements AutoCloseable { public synchronized CompletableFuture<Void> triggerReadToTheEndAndExit() { exitOnEndOfTopic = true; return this.exitFuture; } FunctionAssignmentTailer(
            FunctionRuntimeManager functionRuntimeManager,
            ReaderBuilder readerBuilder,
            WorkerConfig workerConfig,
            ErrorNotifier errorNotifier); synchronized CompletableFuture<Void> triggerReadToTheEndAndExit(); void startFromMessage(MessageId startMessageId); synchronized void start(); @Override synchronized void close(); }
FunctionAssignmentTailer implements AutoCloseable { public synchronized CompletableFuture<Void> triggerReadToTheEndAndExit() { exitOnEndOfTopic = true; return this.exitFuture; } FunctionAssignmentTailer(
            FunctionRuntimeManager functionRuntimeManager,
            ReaderBuilder readerBuilder,
            WorkerConfig workerConfig,
            ErrorNotifier errorNotifier); synchronized CompletableFuture<Void> triggerReadToTheEndAndExit(); void startFromMessage(MessageId startMessageId); synchronized void start(); @Override synchronized void close();  }