test:: test parent link with multiple children

crates/fluvio-stream-dispatcher/src/metadata/local.rs

@@ -1326,6 +1326,66 @@ spec:
                 drop(meta_folder)
             }
 
+            #[fluvio_future::test]
+            async fn test_parent_link_with_multiple_children() {
+                //given
+                let meta_folder = tempfile::tempdir().expect("temp dir created");
+                let meta_store = LocalMetadataStorage::new(&meta_folder);
+                let (mut parent, mut children) = test_parent_with_children(4);
+
+                let child = children.remove(0);
+                let child2 = children.remove(0);
+                let child3 = children.remove(0);
+                let child4 = children.remove(0);
+
+                // only parent without children
+                parent.ctx_mut().item_mut().set_children(Default::default());
+                meta_store
+                    .apply(parent.clone())
+                    .await
+                    .expect("applied parent");
+
+                //when apllying children
+                let (result1, result2, result3, result4) = tokio::join!(
+                    meta_store.apply(child.clone()),
+                    meta_store.apply(child2.clone()),
+                    meta_store.apply(child3.clone()),
+                    meta_store.apply(child4.clone())
+                );
+
+                result1.expect("applied child");
+                result2.expect("applied child");
+                result3.expect("applied child");
+                result4.expect("applied child");
+
+
+                let parent_meta = meta_store
+                    .retrieve_items::<ParentSpec>(&NameSpace::All)
+                    .await
+                    .expect("items")
+                    .items
+                    .remove(0)
+                    .ctx_owned()
+                    .into_inner();
+
+
+                //then
+                assert_eq!(parent_meta.children().unwrap().len(), 1);
+                let children = parent_meta
+                    .children()
+                    .unwrap()
+                    .get(TestSpec::LABEL)
+                    .expect("test spec children");
+
+                assert_eq!(children.len(), 4);
+                assert!(children.iter().any(|c| c.id == child.ctx.item().id));
+                assert!(children.iter().any(|c| c.id == child2.ctx.item().id));
+                assert!(children.iter().any(|c| c.id == child3.ctx.item().id));
+                assert!(children.iter().any(|c| c.id == child4.ctx.item().id));
+                drop(meta_folder)
+            }
+
+
             #[fluvio_future::test]
             async fn test_parent_is_not_existed() {
                 //given