public void notifyKvStateUnregistered(
            JobVertexID jobVertexId, KeyGroupRange keyGroupRange, String registrationName) {

        KvStateLocation location = lookupTable.get(registrationName);

        if (location != null) {
            // Duplicate name if vertex IDs don't match
            if (!location.getJobVertexId().equals(jobVertexId)) {
                throw new IllegalArgumentException(
                        "Another operator ("
                                + location.getJobVertexId()
                                + ") registered the KvState "
                                + "under '"
                                + registrationName
                                + "'.");
            }

            location.unregisterKvState(keyGroupRange);

            if (location.getNumRegisteredKeyGroups() == 0) {
                lookupTable.remove(registrationName);
            }
        } else {
            throw new IllegalArgumentException(
                    "Unknown registration name '"
                            + registrationName
                            + "'. "
                            + "Probably registration/unregistration race.");
        }
    }
@Test
    void testUnregisterBeforeRegister() throws Exception {
        ExecutionJobVertex vertex = createJobVertex(4);
        Map<JobVertexID, ExecutionJobVertex> vertexMap = createVertexMap(vertex);

        KvStateLocationRegistry registry = new KvStateLocationRegistry(new JobID(), vertexMap);
        assertThatThrownBy(
                        () ->
                                registry.notifyKvStateUnregistered(
                                        vertex.getJobVertexId(),
                                        new KeyGroupRange(0, 0),
                                        "any-name"))
                .withFailMessage(
                        "Did not throw expected Exception, because of missing registration")
                .isInstanceOf(IllegalArgumentException.class);
    }