public synchronized CompletableFuture<Void> triggerReadToTheEndAndExit() {
        exitOnEndOfTopic = true;
        return this.exitFuture;
    }
@Test(timeOut = 10000)
    public void testTriggerReadToTheEndAndExit() throws Exception {
        WorkerConfig workerConfig = new WorkerConfig();
        workerConfig.setWorkerId("worker-1");
        workerConfig.setFunctionRuntimeFactoryClassName(ThreadRuntimeFactory.class.getName());
        workerConfig.setFunctionRuntimeFactoryConfigs(
                ObjectMapperFactory.getMapper().getObjectMapper().convertValue(
                        new ThreadRuntimeFactoryConfig().setThreadGroupName("test"), Map.class));
        workerConfig.setPulsarServiceUrl("pulsar://localhost:6650");
        workerConfig.setStateStorageServiceUrl("foo");
        workerConfig.setFunctionAssignmentTopicName("assignments");
        workerConfig.setPulsarFunctionsCluster(CLUSTER_NAME);

        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);

        MessageMetadata metadata = new MessageMetadata();
        Message message1 = spy(new MessageImpl("foo", messageId1.toString(),
                new HashMap<>(), Unpooled.copiedBuffer(assignment1.toByteArray()), null, metadata));
        doReturn(FunctionCommon.getFullyQualifiedInstanceId(assignment1.getInstance())).when(message1).getKey();

        Message message2 = spy(new MessageImpl("foo", messageId2.toString(),
                new HashMap<>(), Unpooled.copiedBuffer(assignment2.toByteArray()), null, metadata));
        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();
        PulsarWorkerService workerService = mock(PulsarWorkerService.class);
        doReturn(pulsarClient).when(workerService).getClient();
        doReturn(mock(PulsarAdmin.class)).when(workerService).getFunctionAdmin();

        ErrorNotifier errorNotifier = spy(ErrorNotifier.getDefaultImpl());

        // test new assignment add functions
        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();
    }