@Override
    public void updateMember(ShareGroupMember newMember) {
        if (newMember == null) {
            throw new IllegalArgumentException("newMember cannot be null.");
        }

        ShareGroupMember oldMember = members.put(newMember.memberId(), newMember);
        maybeUpdateSubscribedTopicNamesAndGroupSubscriptionType(oldMember, newMember);
        maybeUpdateGroupState();
    }
@Test
    public void testUpdateSubscriptionMetadata() {
        Uuid fooTopicId = Uuid.randomUuid();
        Uuid barTopicId = Uuid.randomUuid();
        Uuid zarTopicId = Uuid.randomUuid();

        MetadataImage image = new MetadataImageBuilder()
            .addTopic(fooTopicId, "foo", 1)
            .addTopic(barTopicId, "bar", 2)
            .addTopic(zarTopicId, "zar", 3)
            .addRacks()
            .build();

        ShareGroupMember member1 = new ShareGroupMember.Builder("member1")
            .setSubscribedTopicNames(Collections.singletonList("foo"))
            .build();
        ShareGroupMember member2 = new ShareGroupMember.Builder("member2")
            .setSubscribedTopicNames(Collections.singletonList("bar"))
            .build();
        ShareGroupMember member3 = new ShareGroupMember.Builder("member3")
            .setSubscribedTopicNames(Collections.singletonList("zar"))
            .build();

        ShareGroup shareGroup = createShareGroup("group-foo");

        // It should be empty by default.
        assertEquals(
            Collections.emptyMap(),
            shareGroup.computeSubscriptionMetadata(
                shareGroup.computeSubscribedTopicNames(null, null),
                image.topics(),
                image.cluster()
            )
        );

        // Compute while taking into account member 1.
        assertEquals(
            mkMap(
                mkEntry("foo", new TopicMetadata(fooTopicId, "foo", 1, mkMapOfPartitionRacks(1)))
            ),
            shareGroup.computeSubscriptionMetadata(
                shareGroup.computeSubscribedTopicNames(null, member1),
                image.topics(),
                image.cluster()
            )
        );

        // Updating the group with member1.
        shareGroup.updateMember(member1);

        // It should return foo now.
        assertEquals(
            mkMap(
                mkEntry("foo", new TopicMetadata(fooTopicId, "foo", 1, mkMapOfPartitionRacks(1)))
            ),
            shareGroup.computeSubscriptionMetadata(
                shareGroup.computeSubscribedTopicNames(null, null),
                image.topics(),
                image.cluster()
            )
        );

        // Compute while taking into account removal of member 1.
        assertEquals(
            Collections.emptyMap(),
            shareGroup.computeSubscriptionMetadata(
                shareGroup.computeSubscribedTopicNames(member1, null),
                image.topics(),
                image.cluster()
            )
        );

        // Compute while taking into account member 2.
        assertEquals(
            mkMap(
                mkEntry("foo", new TopicMetadata(fooTopicId, "foo", 1, mkMapOfPartitionRacks(1))),
                mkEntry("bar", new TopicMetadata(barTopicId, "bar", 2, mkMapOfPartitionRacks(2)))
            ),
            shareGroup.computeSubscriptionMetadata(
                shareGroup.computeSubscribedTopicNames(null, member2),
                image.topics(),
                image.cluster()
            )
        );

        // Updating the group with member2.
        shareGroup.updateMember(member2);

        // It should return foo and bar.
        assertEquals(
            mkMap(
                mkEntry("foo", new TopicMetadata(fooTopicId, "foo", 1, mkMapOfPartitionRacks(1))),
                mkEntry("bar", new TopicMetadata(barTopicId, "bar", 2, mkMapOfPartitionRacks(2)))
            ),
            shareGroup.computeSubscriptionMetadata(
                shareGroup.computeSubscribedTopicNames(null, null),
                image.topics(),
                image.cluster()
            )
        );

        // Compute while taking into account removal of member 2.
        assertEquals(
            mkMap(
                mkEntry("foo", new TopicMetadata(fooTopicId, "foo", 1, mkMapOfPartitionRacks(1)))
            ),
            shareGroup.computeSubscriptionMetadata(
                shareGroup.computeSubscribedTopicNames(member2, null),
                image.topics(),
                image.cluster()
            )
        );

        // Removing member1 results in returning bar.
        assertEquals(
            mkMap(
                mkEntry("bar", new TopicMetadata(barTopicId, "bar", 2, mkMapOfPartitionRacks(2)))
            ),
            shareGroup.computeSubscriptionMetadata(
                shareGroup.computeSubscribedTopicNames(member1, null),
                image.topics(),
                image.cluster()
            )
        );

        // Compute while taking into account member 3.
        assertEquals(
            mkMap(
                mkEntry("foo", new TopicMetadata(fooTopicId, "foo", 1, mkMapOfPartitionRacks(1))),
                mkEntry("bar", new TopicMetadata(barTopicId, "bar", 2, mkMapOfPartitionRacks(2))),
                mkEntry("zar", new TopicMetadata(zarTopicId, "zar", 3, mkMapOfPartitionRacks(3)))
            ),
            shareGroup.computeSubscriptionMetadata(
                shareGroup.computeSubscribedTopicNames(null, member3),
                image.topics(),
                image.cluster()
            )
        );

        // Updating group with member3.
        shareGroup.updateMember(member3);

        // It should return foo, bar and zar.
        assertEquals(
            mkMap(
                mkEntry("foo", new TopicMetadata(fooTopicId, "foo", 1, mkMapOfPartitionRacks(1))),
                mkEntry("bar", new TopicMetadata(barTopicId, "bar", 2, mkMapOfPartitionRacks(2))),
                mkEntry("zar", new TopicMetadata(zarTopicId, "zar", 3, mkMapOfPartitionRacks(3)))
            ),
            shareGroup.computeSubscriptionMetadata(
                shareGroup.computeSubscribedTopicNames(null, null),
                image.topics(),
                image.cluster()
            )
        );

        // Compute while taking into account removal of member 1, member 2 and member 3
        assertEquals(
            Collections.emptyMap(),
            shareGroup.computeSubscriptionMetadata(
                shareGroup.computeSubscribedTopicNames(new HashSet<>(Arrays.asList(member1, member2, member3))),
                image.topics(),
                image.cluster()
            )
        );

        // Compute while taking into account removal of member 2 and member 3.
        assertEquals(
            mkMap(
                mkEntry("foo", new TopicMetadata(fooTopicId, "foo", 1, mkMapOfPartitionRacks(1)))
            ),
            shareGroup.computeSubscriptionMetadata(
                shareGroup.computeSubscribedTopicNames(new HashSet<>(Arrays.asList(member2, member3))),
                image.topics(),
                image.cluster()
            )
        );

        // Compute while taking into account removal of member 1.
        assertEquals(
            mkMap(
                mkEntry("bar", new TopicMetadata(barTopicId, "bar", 2, mkMapOfPartitionRacks(2))),
                mkEntry("zar", new TopicMetadata(zarTopicId, "zar", 3, mkMapOfPartitionRacks(3)))
            ),
            shareGroup.computeSubscriptionMetadata(
                shareGroup.computeSubscribedTopicNames(Collections.singleton(member1)),
                image.topics(),
                image.cluster()
            )
        );

        // It should return foo, bar and zar.
        assertEquals(
            mkMap(
                mkEntry("foo", new TopicMetadata(fooTopicId, "foo", 1, mkMapOfPartitionRacks(1))),
                mkEntry("bar", new TopicMetadata(barTopicId, "bar", 2, mkMapOfPartitionRacks(2))),
                mkEntry("zar", new TopicMetadata(zarTopicId, "zar", 3, mkMapOfPartitionRacks(3)))
            ),
            shareGroup.computeSubscriptionMetadata(
                shareGroup.computeSubscribedTopicNames(Collections.emptySet()),
                image.topics(),
                image.cluster()
            )
        );
    }