Fix KV store for group-to-message relationship

The `this.messagePrefix + groupId + '_' + messageId` logic
is introduced since `6.2` via: 64f8ed5

The current fix for new `getMessageFromGroup()` and `removeMessageFromGroupById()`
is to retrieve group and then perform plain `get` and `delete` against just message id

Related to: #8773

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java

@@ -269,13 +269,19 @@ public void removeMessagesFromGroup(Object groupId, Collection<Message<?>> messa
 	public Message<?> getMessageFromGroup(Object groupId, UUID messageId) {
 		Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
 		Assert.notNull(messageId, "'messageId' must not be null");
-		Object object = doRetrieve(this.messagePrefix + groupId + '_' + messageId);
-		if (object != null) {
-			return extractMessage(object);
-		}
-		else {
-			return null;
+
+		Object mgm = doRetrieve(this.groupPrefix + groupId);
+		if (mgm != null) {
+			Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
+			MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
+
+			for (UUID id : messageGroupMetadata.getMessageIds()) {
+				if (id.equals(messageId)) {
+					return getMessage(messageId);
+				}
+			}
 		}
+		return null;
 	}
 
 	@Override
@@ -289,8 +295,7 @@ public boolean removeMessageFromGroupById(Object groupId, UUID messageId) {
 
 			if (messageGroupMetadata.getMessageIds().contains(messageId)) {
 				messageGroupMetadata.remove(messageId);
-				String groupToMessageId = this.messagePrefix + groupId + '_' + messageId;
-				if (doRemove(groupToMessageId) != null) {
+				if (removeMessage(messageId) != null) {
 					messageGroupMetadata.setLastModified(System.currentTimeMillis());
 					doStore(this.groupPrefix + groupId, messageGroupMetadata);
 					return true;