opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/sharding/DistributedShardedDOMDataTreeTest.java

   1 /*
   2  * Copyright (c) 2016 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.cluster.sharding;
  10
  11 import static org.junit.Assert.assertEquals;
  12 import static org.junit.Assert.assertNotNull;
  13 import static org.junit.Assert.assertTrue;
  14 import static org.mockito.Matchers.anyCollection;
  15 import static org.mockito.Matchers.anyMap;
  16 import static org.mockito.Mockito.doNothing;
  17 import static org.mockito.Mockito.doReturn;
  18 import static org.mockito.Mockito.mock;
  19 import static org.mockito.Mockito.timeout;
  20 import static org.mockito.Mockito.verify;
  21 import static org.mockito.Mockito.verifyNoMoreInteractions;
  22 import static org.opendaylight.controller.cluster.datastore.IntegrationTestKit.findLocalShard;
  23 import static org.opendaylight.controller.cluster.datastore.IntegrationTestKit.waitUntilShardIsDown;
  24
  25 import akka.actor.ActorRef;
  26 import akka.actor.ActorSystem;
  27 import akka.actor.Address;
  28 import akka.actor.AddressFromURIString;
  29 import akka.actor.Props;
  30 import akka.cluster.Cluster;
  31 import akka.testkit.JavaTestKit;
  32 import com.google.common.base.Optional;
  33 import com.google.common.collect.Lists;
  34 import com.google.common.util.concurrent.CheckedFuture;
  35 import com.typesafe.config.ConfigFactory;
  36 import java.util.ArrayList;
  37 import java.util.Collection;
  38 import java.util.Collections;
  39 import java.util.List;
  40 import java.util.Map;
  41 import java.util.Set;
  42 import java.util.concurrent.CompletionStage;
  43 import java.util.concurrent.TimeUnit;
  44 import org.junit.After;
  45 import org.junit.Assert;
  46 import org.junit.Before;
  47 import org.junit.Ignore;
  48 import org.junit.Test;
  49 import org.mockito.ArgumentCaptor;
  50 import org.mockito.Captor;
  51 import org.mockito.Mockito;
  52 import org.mockito.MockitoAnnotations;
  53 import org.opendaylight.controller.cluster.ActorSystemProvider;
  54 import org.opendaylight.controller.cluster.access.concepts.MemberName;
  55 import org.opendaylight.controller.cluster.databroker.actors.dds.ClientLocalHistory;
  56 import org.opendaylight.controller.cluster.databroker.actors.dds.ClientTransaction;
  57 import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
  58 import org.opendaylight.controller.cluster.databroker.actors.dds.SimpleDataStoreClientActor;
  59 import org.opendaylight.controller.cluster.datastore.AbstractTest;
  60 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
  61 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
  62 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
  63 import org.opendaylight.controller.cluster.datastore.IntegrationTestKit;
  64 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  65 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
  66 import org.opendaylight.controller.cluster.dom.api.CDSDataTreeProducer;
  67 import org.opendaylight.controller.cluster.dom.api.CDSShardAccess;
  68 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
  69 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
  70 import org.opendaylight.controller.cluster.sharding.DistributedShardFactory.DistributedShardRegistration;
  71 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  72 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  73 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  74 import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
  75 import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
  76 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  77 import org.opendaylight.mdsal.dom.api.DOMDataTreeListener;
  78 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer;
  79 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
  80 import org.opendaylight.yangtools.yang.common.QName;
  81 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  82 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  83 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
  84 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  85 import org.opendaylight.yangtools.yang.data.api.schema.LeafNode;
  86 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  87 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
  88 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  89 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  90 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  91 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
  92 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafNodeBuilder;
  93 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableMapNodeBuilder;
  94 import org.slf4j.Logger;
  95 import org.slf4j.LoggerFactory;
  96
  97 public class DistributedShardedDOMDataTreeTest extends AbstractTest {
  98
  99     private static final Logger LOG = LoggerFactory.getLogger(DistributedShardedDOMDataTreeRemotingTest.class);
 100
 101     private static final Address MEMBER_1_ADDRESS =
 102             AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558");
 103
 104     private static final DOMDataTreeIdentifier TEST_ID =
 105             new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH);
 106
 107     private static final DOMDataTreeIdentifier INNER_LIST_ID =
 108             new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION,
 109                     YangInstanceIdentifier.create(getOuterListIdFor(0).getPathArguments())
 110                             .node(TestModel.INNER_LIST_QNAME));
 111     private static final Set<MemberName> SINGLE_MEMBER = Collections.singleton(AbstractTest.MEMBER_NAME);
 112
 113     private static final String MODULE_SHARDS_CONFIG = "module-shards-default-member-1.conf";
 114
 115     private ActorSystem leaderSystem;
 116
 117     private final Builder leaderDatastoreContextBuilder =
 118             DatastoreContext.newBuilder()
 119                     .shardHeartbeatIntervalInMillis(100)
 120                     .shardElectionTimeoutFactor(2)
 121                     .logicalStoreType(
 122                             org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType.CONFIGURATION);
 123
 124     private DistributedDataStore leaderDistributedDataStore;
 125     private DistributedDataStore operDistributedDatastore;
 126     private IntegrationTestKit leaderTestKit;
 127
 128     private DistributedShardedDOMDataTree leaderShardFactory;
 129
 130     @Captor
 131     private ArgumentCaptor<Collection<DataTreeCandidate>> captorForChanges;
 132     @Captor
 133     private ArgumentCaptor<Map<DOMDataTreeIdentifier, NormalizedNode<?, ?>>> captorForSubtrees;
 134
 135     private ActorSystemProvider leaderSystemProvider;
 136
 137     @Before
 138     public void setUp() {
 139         MockitoAnnotations.initMocks(this);
 140
 141         InMemoryJournal.clear();
 142         InMemorySnapshotStore.clear();
 143
 144         leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
 145         Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS);
 146
 147         leaderSystemProvider = Mockito.mock(ActorSystemProvider.class);
 148         doReturn(leaderSystem).when(leaderSystemProvider).getActorSystem();
 149     }
 150
 151     @After
 152     public void tearDown() throws Exception {
 153         if (leaderDistributedDataStore != null) {
 154             leaderDistributedDataStore.close();
 155         }
 156
 157         if (operDistributedDatastore != null) {
 158             operDistributedDatastore.close();
 159         }
 160
 161         JavaTestKit.shutdownActorSystem(leaderSystem);
 162
 163         InMemoryJournal.clear();
 164         InMemorySnapshotStore.clear();
 165     }
 166
 167     private void initEmptyDatastores() throws Exception {
 168         leaderTestKit = new IntegrationTestKit(leaderSystem, leaderDatastoreContextBuilder);
 169
 170         leaderDistributedDataStore = leaderTestKit.setupDistributedDataStore(
 171                 "config", MODULE_SHARDS_CONFIG, "empty-modules.conf", true,
 172                 SchemaContextHelper.distributedShardedDOMDataTreeSchemaContext());
 173
 174         operDistributedDatastore = leaderTestKit.setupDistributedDataStore(
 175                 "operational", MODULE_SHARDS_CONFIG, "empty-modules.conf",true,
 176                 SchemaContextHelper.distributedShardedDOMDataTreeSchemaContext());
 177
 178         leaderShardFactory = new DistributedShardedDOMDataTree(leaderSystemProvider,
 179                 operDistributedDatastore,
 180                 leaderDistributedDataStore);
 181
 182         leaderShardFactory.init();
 183     }
 184
 185
 186     @Test
 187     public void testWritesIntoDefaultShard() throws Exception {
 188         initEmptyDatastores();
 189
 190         final DOMDataTreeIdentifier configRoot =
 191                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY);
 192
 193         final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(configRoot));
 194
 195         final DOMDataTreeCursorAwareTransaction tx = producer.createTransaction(true);
 196         final DOMDataTreeWriteCursor cursor =
 197                 tx.createCursor(new DOMDataTreeIdentifier(
 198                         LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY));
 199         Assert.assertNotNull(cursor);
 200
 201         final ContainerNode test =
 202                 ImmutableContainerNodeBuilder.create()
 203                         .withNodeIdentifier(new NodeIdentifier(TestModel.TEST_QNAME)).build();
 204
 205         cursor.write(test.getIdentifier(), test);
 206         cursor.close();
 207
 208         tx.submit().checkedGet();
 209     }
 210
 211     @Test
 212     public void testSingleNodeWritesAndRead() throws Exception {
 213         initEmptyDatastores();
 214
 215         final DistributedShardRegistration shardRegistration = waitOnAsyncTask(
 216                 leaderShardFactory.createDistributedShard(TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
 217                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 218
 219         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 220                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
 221
 222         final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
 223
 224         final DOMDataTreeCursorAwareTransaction tx = producer.createTransaction(true);
 225         final DOMDataTreeWriteCursor cursor = tx.createCursor(TEST_ID);
 226         Assert.assertNotNull(cursor);
 227         final YangInstanceIdentifier nameId =
 228                 YangInstanceIdentifier.builder(TestModel.TEST_PATH).node(TestModel.NAME_QNAME).build();
 229         final LeafNode<String> valueToCheck = ImmutableLeafNodeBuilder.<String>create().withNodeIdentifier(
 230                 new NodeIdentifier(TestModel.NAME_QNAME)).withValue("Test Value").build();
 231         LOG.debug("Writing data {} at {}, cursor {}", nameId.getLastPathArgument(), valueToCheck, cursor);
 232         cursor.write(nameId.getLastPathArgument(),
 233                 valueToCheck);
 234
 235         cursor.close();
 236         LOG.debug("Got to pre submit");
 237
 238         tx.submit().checkedGet();
 239
 240         final DOMDataTreeListener mockedDataTreeListener = mock(DOMDataTreeListener.class);
 241         doNothing().when(mockedDataTreeListener).onDataTreeChanged(anyCollection(), anyMap());
 242
 243         leaderShardFactory.registerListener(mockedDataTreeListener, Collections.singletonList(TEST_ID),
 244                 true, Collections.emptyList());
 245
 246         verify(mockedDataTreeListener, timeout(1000).times(1)).onDataTreeChanged(captorForChanges.capture(),
 247                 captorForSubtrees.capture());
 248         final List<Collection<DataTreeCandidate>> capturedValue = captorForChanges.getAllValues();
 249
 250         final java.util.Optional<NormalizedNode<?, ?>> dataAfter =
 251                 capturedValue.get(0).iterator().next().getRootNode().getDataAfter();
 252
 253         final NormalizedNode<?,?> expected = ImmutableContainerNodeBuilder.create()
 254                 .withNodeIdentifier(new NodeIdentifier(TestModel.TEST_QNAME)).withChild(valueToCheck).build();
 255         assertEquals(expected, dataAfter.get());
 256
 257         verifyNoMoreInteractions(mockedDataTreeListener);
 258
 259         final String shardName = ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier());
 260         LOG.debug("Creating distributed datastore client for shard {}", shardName);
 261
 262         final ActorContext actorContext = leaderDistributedDataStore.getActorContext();
 263         final Props distributedDataStoreClientProps =
 264                 SimpleDataStoreClientActor.props(actorContext.getCurrentMemberName(),
 265                         "Shard-" + shardName, actorContext, shardName);
 266
 267         final ActorRef clientActor = leaderSystem.actorOf(distributedDataStoreClientProps);
 268         final DataStoreClient distributedDataStoreClient = SimpleDataStoreClientActor
 269                     .getDistributedDataStoreClient(clientActor, 30, TimeUnit.SECONDS);
 270
 271         final ClientLocalHistory localHistory = distributedDataStoreClient.createLocalHistory();
 272         final ClientTransaction tx2 = localHistory.createTransaction();
 273         final CheckedFuture<Optional<NormalizedNode<?, ?>>,
 274                 org.opendaylight.mdsal.common.api.ReadFailedException> read =
 275                 tx2.read(YangInstanceIdentifier.EMPTY);
 276
 277         final Optional<NormalizedNode<?, ?>> optional = read.checkedGet();
 278         tx2.abort();
 279         localHistory.close();
 280
 281         shardRegistration.close().toCompletableFuture().get();
 282
 283     }
 284
 285     @Test
 286     public void testMultipleWritesIntoSingleMapEntry() throws Exception {
 287         initEmptyDatastores();
 288
 289         final DistributedShardRegistration shardRegistration = waitOnAsyncTask(
 290                 leaderShardFactory.createDistributedShard(TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
 291                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 292
 293         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 294                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
 295
 296         LOG.warn("Got after waiting for nonleader");
 297         final ActorRef leaderShardManager = leaderDistributedDataStore.getActorContext().getShardManager();
 298
 299         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 300                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 301
 302         final YangInstanceIdentifier oid1 = getOuterListIdFor(0);
 303         final DOMDataTreeIdentifier outerListPath = new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, oid1);
 304
 305         final DistributedShardRegistration outerListShardReg = waitOnAsyncTask(
 306                 leaderShardFactory.createDistributedShard(outerListPath, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
 307                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 308
 309         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 310                 ClusterUtils.getCleanShardName(outerListPath.getRootIdentifier()));
 311
 312         final DOMDataTreeProducer shardProducer = leaderShardFactory.createProducer(
 313                 Collections.singletonList(outerListPath));
 314
 315         final DOMDataTreeCursorAwareTransaction tx = shardProducer.createTransaction(false);
 316         final DOMDataTreeWriteCursor cursor =
 317                 tx.createCursor(new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, oid1));
 318         assertNotNull(cursor);
 319
 320         MapNode innerList = ImmutableMapNodeBuilder
 321                 .create()
 322                 .withNodeIdentifier(new NodeIdentifier(TestModel.INNER_LIST_QNAME))
 323                 .build();
 324
 325         cursor.write(new NodeIdentifier(TestModel.INNER_LIST_QNAME), innerList);
 326         cursor.close();
 327         tx.submit().checkedGet();
 328
 329         final ArrayList<CheckedFuture<Void, TransactionCommitFailedException>> futures = new ArrayList<>();
 330         for (int i = 0; i < 1000; i++) {
 331             final Collection<MapEntryNode> innerListMapEntries = createInnerListMapEntries(1000, "run-" + i);
 332             for (final MapEntryNode innerListMapEntry : innerListMapEntries) {
 333                 final DOMDataTreeCursorAwareTransaction tx1 = shardProducer.createTransaction(false);
 334                 final DOMDataTreeWriteCursor cursor1 = tx1.createCursor(
 335                         new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION,
 336                                 oid1.node(new NodeIdentifier(TestModel.INNER_LIST_QNAME))));
 337                 cursor1.write(innerListMapEntry.getIdentifier(), innerListMapEntry);
 338                 cursor1.close();
 339                 futures.add(tx1.submit());
 340             }
 341         }
 342
 343         futures.get(futures.size() - 1).checkedGet();
 344
 345         final DOMDataTreeListener mockedDataTreeListener = mock(DOMDataTreeListener.class);
 346         doNothing().when(mockedDataTreeListener).onDataTreeChanged(anyCollection(), anyMap());
 347
 348         leaderShardFactory.registerListener(mockedDataTreeListener, Collections.singletonList(INNER_LIST_ID),
 349                 true, Collections.emptyList());
 350
 351         verify(mockedDataTreeListener, timeout(1000).times(1)).onDataTreeChanged(captorForChanges.capture(),
 352                 captorForSubtrees.capture());
 353         verifyNoMoreInteractions(mockedDataTreeListener);
 354         final List<Collection<DataTreeCandidate>> capturedValue = captorForChanges.getAllValues();
 355
 356         final NormalizedNode<?,?> expected =
 357                 ImmutableMapNodeBuilder
 358                         .create()
 359                         .withNodeIdentifier(new NodeIdentifier(TestModel.INNER_LIST_QNAME))
 360                                 // only the values from the last run should be present
 361                         .withValue(createInnerListMapEntries(1000, "run-999"))
 362                         .build();
 363
 364         assertEquals("List values dont match the expected values from the last run",
 365                 expected, capturedValue.get(0).iterator().next().getRootNode().getDataAfter().get());
 366
 367     }
 368
 369     // top level shard at TEST element, with subshards on each outer-list map entry
 370     @Test
 371     @Ignore
 372     public void testMultipleShardLevels() throws Exception {
 373         initEmptyDatastores();
 374
 375         final DistributedShardRegistration testShardReg = waitOnAsyncTask(
 376                 leaderShardFactory.createDistributedShard(TEST_ID, SINGLE_MEMBER),
 377                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 378
 379         final ArrayList<DistributedShardRegistration> registrations = new ArrayList<>();
 380         final int listSize = 5;
 381         for (int i = 0; i < listSize; i++) {
 382             final YangInstanceIdentifier entryYID = getOuterListIdFor(i);
 383             final CompletionStage<DistributedShardRegistration> future = leaderShardFactory.createDistributedShard(
 384                     new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, entryYID), SINGLE_MEMBER);
 385
 386             registrations.add(waitOnAsyncTask(future, DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION));
 387         }
 388
 389         final DOMDataTreeIdentifier rootId =
 390                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY);
 391         final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singletonList(
 392                 rootId));
 393
 394         DOMDataTreeCursorAwareTransaction transaction = producer.createTransaction(false);
 395
 396         DOMDataTreeWriteCursor cursor = transaction.createCursor(rootId);
 397         assertNotNull(cursor);
 398
 399         final MapNode outerList =
 400                 ImmutableMapNodeBuilder.create()
 401                         .withNodeIdentifier(new NodeIdentifier(TestModel.OUTER_LIST_QNAME)).build();
 402
 403         final ContainerNode testNode =
 404                 ImmutableContainerNodeBuilder.create()
 405                         .withNodeIdentifier(new NodeIdentifier(TestModel.TEST_QNAME))
 406                         .withChild(outerList)
 407                         .build();
 408
 409         cursor.write(testNode.getIdentifier(), testNode);
 410
 411         cursor.close();
 412         transaction.submit().checkedGet();
 413
 414         final DOMDataTreeListener mockedDataTreeListener = mock(DOMDataTreeListener.class);
 415         doNothing().when(mockedDataTreeListener).onDataTreeChanged(anyCollection(), anyMap());
 416
 417         final MapNode wholeList = ImmutableMapNodeBuilder.create(outerList)
 418                 .withValue(createOuterEntries(listSize, "testing-values")).build();
 419
 420         transaction = producer.createTransaction(false);
 421         cursor = transaction.createCursor(TEST_ID);
 422         assertNotNull(cursor);
 423
 424         cursor.write(wholeList.getIdentifier(), wholeList);
 425         cursor.close();
 426
 427         transaction.submit().checkedGet();
 428
 429         leaderShardFactory.registerListener(mockedDataTreeListener, Collections.singletonList(TEST_ID),
 430                 true, Collections.emptyList());
 431
 432         verify(mockedDataTreeListener, timeout(35000).atLeast(2)).onDataTreeChanged(captorForChanges.capture(),
 433                 captorForSubtrees.capture());
 434         verifyNoMoreInteractions(mockedDataTreeListener);
 435         final List<Map<DOMDataTreeIdentifier, NormalizedNode<?, ?>>> allSubtrees = captorForSubtrees.getAllValues();
 436
 437         final Map<DOMDataTreeIdentifier, NormalizedNode<?, ?>> lastSubtree = allSubtrees.get(allSubtrees.size() - 1);
 438
 439         final NormalizedNode<?, ?> actual = lastSubtree.get(TEST_ID);
 440         assertNotNull(actual);
 441
 442         final NormalizedNode<?, ?> expected =
 443                 ImmutableContainerNodeBuilder.create()
 444                         .withNodeIdentifier(new NodeIdentifier(TestModel.TEST_QNAME))
 445                         .withChild(ImmutableMapNodeBuilder.create(outerList)
 446                                 .withValue(createOuterEntries(listSize, "testing-values")).build())
 447                         .build();
 448
 449
 450         for (final DistributedShardRegistration registration : registrations) {
 451             waitOnAsyncTask(registration.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 452         }
 453
 454         waitOnAsyncTask(testShardReg.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 455
 456         assertEquals(expected, actual);
 457     }
 458
 459     @Test
 460     public void testMultipleRegistrationsAtOnePrefix() throws Exception {
 461         initEmptyDatastores();
 462
 463         for (int i = 0; i < 10; i++) {
 464             LOG.debug("Round {}", i);
 465             final DistributedShardRegistration reg1 = waitOnAsyncTask(leaderShardFactory.createDistributedShard(
 466                     TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
 467                     DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 468
 469             leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 470                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 471
 472             assertNotNull(findLocalShard(leaderDistributedDataStore.getActorContext(),
 473                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
 474
 475             waitOnAsyncTask(reg1.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 476
 477             waitUntilShardIsDown(leaderDistributedDataStore.getActorContext(),
 478                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 479         }
 480     }
 481
 482     @Test
 483     public void testCDSDataTreeProducer() throws Exception {
 484         initEmptyDatastores();
 485
 486         final DistributedShardRegistration reg1 = waitOnAsyncTask(leaderShardFactory.createDistributedShard(
 487                 TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
 488                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 489
 490         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 491                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 492
 493         assertNotNull(findLocalShard(leaderDistributedDataStore.getActorContext(),
 494                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
 495
 496
 497         final DOMDataTreeIdentifier configRoot =
 498                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY);
 499         final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(configRoot));
 500
 501         assertTrue(producer instanceof CDSDataTreeProducer);
 502
 503         final CDSDataTreeProducer cdsProducer = (CDSDataTreeProducer) producer;
 504         CDSShardAccess shardAccess = cdsProducer.getShardAccess(TEST_ID);
 505         assertEquals(shardAccess.getShardIdentifier(), TEST_ID);
 506
 507         shardAccess = cdsProducer.getShardAccess(INNER_LIST_ID);
 508         assertEquals(TEST_ID, shardAccess.getShardIdentifier());
 509
 510         shardAccess = cdsProducer.getShardAccess(configRoot);
 511         assertEquals(configRoot, shardAccess.getShardIdentifier());
 512
 513         waitOnAsyncTask(reg1.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 514     }
 515
 516     private static Collection<MapEntryNode> createOuterEntries(final int amount, final String valuePrefix) {
 517         final Collection<MapEntryNode> ret = new ArrayList<>();
 518         for (int i = 0; i < amount; i++) {
 519             ret.add(ImmutableNodes.mapEntryBuilder()
 520                     .withNodeIdentifier(new NodeIdentifierWithPredicates(TestModel.OUTER_LIST_QNAME,
 521                             QName.create(TestModel.OUTER_LIST_QNAME, "id"), i))
 522                     .withChild(ImmutableNodes
 523                             .leafNode(QName.create(TestModel.OUTER_LIST_QNAME, "id"), i))
 524                     .withChild(createWholeInnerList(amount, "outer id: " + i + " " + valuePrefix))
 525                     .build());
 526         }
 527
 528         return ret;
 529     }
 530
 531     private static MapNode createWholeInnerList(final int amount, final String valuePrefix) {
 532         return ImmutableMapNodeBuilder.create().withNodeIdentifier(new NodeIdentifier(TestModel.INNER_LIST_QNAME))
 533                 .withValue(createInnerListMapEntries(amount, valuePrefix)).build();
 534     }
 535
 536     private static Collection<MapEntryNode> createInnerListMapEntries(final int amount, final String valuePrefix) {
 537         final Collection<MapEntryNode> ret = new ArrayList<>();
 538         for (int i = 0; i < amount; i++) {
 539             ret.add(ImmutableNodes.mapEntryBuilder()
 540                     .withNodeIdentifier(new NodeIdentifierWithPredicates(TestModel.INNER_LIST_QNAME,
 541                             QName.create(TestModel.INNER_LIST_QNAME, "name"), Integer.toString(i)))
 542                     .withChild(ImmutableNodes
 543                             .leafNode(QName.create(TestModel.INNER_LIST_QNAME, "value"), valuePrefix + "-" + i))
 544                     .build());
 545         }
 546
 547         return ret;
 548     }
 549
 550     private static YangInstanceIdentifier getOuterListIdFor(final int id) {
 551         return TestModel.OUTER_LIST_PATH.node(new NodeIdentifierWithPredicates(
 552                 TestModel.OUTER_LIST_QNAME, QName.create(TestModel.OUTER_LIST_QNAME, "id"), id));
 553     }
 554 }