public void notifyKvStateRegistered(
            JobVertexID jobVertexId,
            KeyGroupRange keyGroupRange,
            String registrationName,
            KvStateID kvStateId,
            InetSocketAddress kvStateServerAddress) {

        KvStateLocation location = lookupTable.get(registrationName);

        if (location == null) {
            // First registration for this operator, create the location info
            ExecutionJobVertex vertex = jobVertices.get(jobVertexId);

            if (vertex != null) {
                int parallelism = vertex.getMaxParallelism();
                location = new KvStateLocation(jobId, jobVertexId, parallelism, registrationName);
                lookupTable.put(registrationName, location);
            } else {
                throw new IllegalArgumentException("Unknown JobVertexID " + jobVertexId);
            }
        }

        // Duplicated name if vertex IDs don't match
        if (!location.getJobVertexId().equals(jobVertexId)) {
            IllegalStateException duplicate =
                    new IllegalStateException(
                            "Registration name clash. KvState with name '"
                                    + registrationName
                                    + "' has already been registered by another operator ("
                                    + location.getJobVertexId()
                                    + ").");

            ExecutionJobVertex vertex = jobVertices.get(jobVertexId);
            if (vertex != null) {
                vertex.fail(new SuppressRestartsException(duplicate));
            }

            throw duplicate;
        }
        location.registerKvState(keyGroupRange, kvStateId, kvStateServerAddress);
    }
@Test
    void testRegisterDuplicateName() throws Exception {
        ExecutionJobVertex[] vertices =
                new ExecutionJobVertex[] {createJobVertex(32), createJobVertex(13)};

        Map<JobVertexID, ExecutionJobVertex> vertexMap = createVertexMap(vertices);

        String registrationName = "duplicated-name";
        KvStateLocationRegistry registry = new KvStateLocationRegistry(new JobID(), vertexMap);

        // First operator registers
        registry.notifyKvStateRegistered(
                vertices[0].getJobVertexId(),
                new KeyGroupRange(0, 0),
                registrationName,
                new KvStateID(),
                new InetSocketAddress(InetAddress.getLocalHost(), 12328));

        assertThatThrownBy(
                        () ->
                                // Second operator registers same name
                                registry.notifyKvStateRegistered(
                                        vertices[1].getJobVertexId(),
                                        new KeyGroupRange(0, 0),
                                        registrationName,
                                        new KvStateID(),
                                        new InetSocketAddress(InetAddress.getLocalHost(), 12032)))
                .withFailMessage("Did not throw expected Exception after duplicated name")
                .isInstanceOf(IllegalStateException.class);
    }