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.javadsl.TestKit;
  32 import com.google.common.collect.Lists;
  33 import com.google.common.util.concurrent.CheckedFuture;
  34 import com.google.common.util.concurrent.FluentFuture;
  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.Optional;
  42 import java.util.Set;
  43 import java.util.concurrent.CompletionStage;
  44 import java.util.concurrent.TimeUnit;
  45 import org.junit.After;
  46 import org.junit.Assert;
  47 import org.junit.Before;
  48 import org.junit.Ignore;
  49 import org.junit.Test;
  50 import org.mockito.ArgumentCaptor;
  51 import org.mockito.Captor;
  52 import org.mockito.Mockito;
  53 import org.mockito.MockitoAnnotations;
  54 import org.opendaylight.controller.cluster.ActorSystemProvider;
  55 import org.opendaylight.controller.cluster.access.concepts.MemberName;
  56 import org.opendaylight.controller.cluster.databroker.actors.dds.ClientLocalHistory;
  57 import org.opendaylight.controller.cluster.databroker.actors.dds.ClientTransaction;
  58 import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
  59 import org.opendaylight.controller.cluster.databroker.actors.dds.SimpleDataStoreClientActor;
  60 import org.opendaylight.controller.cluster.datastore.AbstractTest;
  61 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
  62 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
  63 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
  64 import org.opendaylight.controller.cluster.datastore.IntegrationTestKit;
  65 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  66 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
  67 import org.opendaylight.controller.cluster.dom.api.CDSDataTreeProducer;
  68 import org.opendaylight.controller.cluster.dom.api.CDSShardAccess;
  69 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
  70 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
  71 import org.opendaylight.controller.cluster.sharding.DistributedShardFactory.DistributedShardRegistration;
  72 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  73 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  74 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  75 import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
  76 import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
  77 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  78 import org.opendaylight.mdsal.dom.api.DOMDataTreeListener;
  79 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer;
  80 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
  81 import org.opendaylight.yangtools.yang.common.QName;
  82 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  83 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  84 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
  85 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  86 import org.opendaylight.yangtools.yang.data.api.schema.LeafNode;
  87 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  88 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
  89 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  90 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  91 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  92 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
  93 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafNodeBuilder;
  94 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableMapNodeBuilder;
  95 import org.slf4j.Logger;
  96 import org.slf4j.LoggerFactory;
  97
  98 public class DistributedShardedDOMDataTreeTest extends AbstractTest {
  99
 100     private static final Logger LOG = LoggerFactory.getLogger(DistributedShardedDOMDataTreeRemotingTest.class);
 101
 102     private static final Address MEMBER_1_ADDRESS =
 103             AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558");
 104
 105     private static final DOMDataTreeIdentifier TEST_ID =
 106             new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH);
 107
 108     private static final DOMDataTreeIdentifier INNER_LIST_ID =
 109             new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION,
 110                     YangInstanceIdentifier.create(getOuterListIdFor(0).getPathArguments())
 111                             .node(TestModel.INNER_LIST_QNAME));
 112     private static final Set<MemberName> SINGLE_MEMBER = Collections.singleton(AbstractTest.MEMBER_NAME);
 113
 114     private static final String MODULE_SHARDS_CONFIG = "module-shards-default-member-1.conf";
 115
 116     private ActorSystem leaderSystem;
 117
 118     private final Builder leaderDatastoreContextBuilder =
 119             DatastoreContext.newBuilder()
 120                     .shardHeartbeatIntervalInMillis(100)
 121                     .shardElectionTimeoutFactor(2)
 122                     .logicalStoreType(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() {
 153         if (leaderDistributedDataStore != null) {
 154             leaderDistributedDataStore.close();
 155         }
 156
 157         if (operDistributedDatastore != null) {
 158             operDistributedDatastore.close();
 159         }
 160
 161         TestKit.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 FluentFuture<Optional<NormalizedNode<?, ?>>> read = tx2.read(YangInstanceIdentifier.EMPTY);
 274
 275         final Optional<NormalizedNode<?, ?>> optional = read.get();
 276         tx2.abort();
 277         localHistory.close();
 278
 279         shardRegistration.close().toCompletableFuture().get();
 280
 281     }
 282
 283     @Test
 284     public void testMultipleWritesIntoSingleMapEntry() throws Exception {
 285         initEmptyDatastores();
 286
 287         final DistributedShardRegistration shardRegistration = waitOnAsyncTask(
 288                 leaderShardFactory.createDistributedShard(TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
 289                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 290
 291         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 292                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
 293
 294         LOG.warn("Got after waiting for nonleader");
 295         final ActorRef leaderShardManager = leaderDistributedDataStore.getActorContext().getShardManager();
 296
 297         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 298                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 299
 300         final YangInstanceIdentifier oid1 = getOuterListIdFor(0);
 301         final DOMDataTreeIdentifier outerListPath = new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, oid1);
 302
 303         final DistributedShardRegistration outerListShardReg = waitOnAsyncTask(
 304                 leaderShardFactory.createDistributedShard(outerListPath, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
 305                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 306
 307         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 308                 ClusterUtils.getCleanShardName(outerListPath.getRootIdentifier()));
 309
 310         final DOMDataTreeProducer shardProducer = leaderShardFactory.createProducer(
 311                 Collections.singletonList(outerListPath));
 312
 313         final DOMDataTreeCursorAwareTransaction tx = shardProducer.createTransaction(false);
 314         final DOMDataTreeWriteCursor cursor =
 315                 tx.createCursor(new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, oid1));
 316         assertNotNull(cursor);
 317
 318         MapNode innerList = ImmutableMapNodeBuilder
 319                 .create()
 320                 .withNodeIdentifier(new NodeIdentifier(TestModel.INNER_LIST_QNAME))
 321                 .build();
 322
 323         cursor.write(new NodeIdentifier(TestModel.INNER_LIST_QNAME), innerList);
 324         cursor.close();
 325         tx.submit().checkedGet();
 326
 327         final ArrayList<CheckedFuture<Void, TransactionCommitFailedException>> futures = new ArrayList<>();
 328         for (int i = 0; i < 1000; i++) {
 329             final Collection<MapEntryNode> innerListMapEntries = createInnerListMapEntries(1000, "run-" + i);
 330             for (final MapEntryNode innerListMapEntry : innerListMapEntries) {
 331                 final DOMDataTreeCursorAwareTransaction tx1 = shardProducer.createTransaction(false);
 332                 final DOMDataTreeWriteCursor cursor1 = tx1.createCursor(
 333                         new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION,
 334                                 oid1.node(new NodeIdentifier(TestModel.INNER_LIST_QNAME))));
 335                 cursor1.write(innerListMapEntry.getIdentifier(), innerListMapEntry);
 336                 cursor1.close();
 337                 futures.add(tx1.submit());
 338             }
 339         }
 340
 341         futures.get(futures.size() - 1).checkedGet();
 342
 343         final DOMDataTreeListener mockedDataTreeListener = mock(DOMDataTreeListener.class);
 344         doNothing().when(mockedDataTreeListener).onDataTreeChanged(anyCollection(), anyMap());
 345
 346         leaderShardFactory.registerListener(mockedDataTreeListener, Collections.singletonList(INNER_LIST_ID),
 347                 true, Collections.emptyList());
 348
 349         verify(mockedDataTreeListener, timeout(1000).times(1)).onDataTreeChanged(captorForChanges.capture(),
 350                 captorForSubtrees.capture());
 351         verifyNoMoreInteractions(mockedDataTreeListener);
 352         final List<Collection<DataTreeCandidate>> capturedValue = captorForChanges.getAllValues();
 353
 354         final NormalizedNode<?,?> expected =
 355                 ImmutableMapNodeBuilder
 356                         .create()
 357                         .withNodeIdentifier(new NodeIdentifier(TestModel.INNER_LIST_QNAME))
 358                                 // only the values from the last run should be present
 359                         .withValue(createInnerListMapEntries(1000, "run-999"))
 360                         .build();
 361
 362         assertEquals("List values dont match the expected values from the last run",
 363                 expected, capturedValue.get(0).iterator().next().getRootNode().getDataAfter().get());
 364
 365     }
 366
 367     // top level shard at TEST element, with subshards on each outer-list map entry
 368     @Test
 369     @Ignore
 370     public void testMultipleShardLevels() throws Exception {
 371         initEmptyDatastores();
 372
 373         final DistributedShardRegistration testShardReg = waitOnAsyncTask(
 374                 leaderShardFactory.createDistributedShard(TEST_ID, SINGLE_MEMBER),
 375                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 376
 377         final ArrayList<DistributedShardRegistration> registrations = new ArrayList<>();
 378         final int listSize = 5;
 379         for (int i = 0; i < listSize; i++) {
 380             final YangInstanceIdentifier entryYID = getOuterListIdFor(i);
 381             final CompletionStage<DistributedShardRegistration> future = leaderShardFactory.createDistributedShard(
 382                     new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, entryYID), SINGLE_MEMBER);
 383
 384             registrations.add(waitOnAsyncTask(future, DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION));
 385         }
 386
 387         final DOMDataTreeIdentifier rootId =
 388                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY);
 389         final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singletonList(
 390                 rootId));
 391
 392         DOMDataTreeCursorAwareTransaction transaction = producer.createTransaction(false);
 393
 394         DOMDataTreeWriteCursor cursor = transaction.createCursor(rootId);
 395         assertNotNull(cursor);
 396
 397         final MapNode outerList =
 398                 ImmutableMapNodeBuilder.create()
 399                         .withNodeIdentifier(new NodeIdentifier(TestModel.OUTER_LIST_QNAME)).build();
 400
 401         final ContainerNode testNode =
 402                 ImmutableContainerNodeBuilder.create()
 403                         .withNodeIdentifier(new NodeIdentifier(TestModel.TEST_QNAME))
 404                         .withChild(outerList)
 405                         .build();
 406
 407         cursor.write(testNode.getIdentifier(), testNode);
 408
 409         cursor.close();
 410         transaction.submit().checkedGet();
 411
 412         final DOMDataTreeListener mockedDataTreeListener = mock(DOMDataTreeListener.class);
 413         doNothing().when(mockedDataTreeListener).onDataTreeChanged(anyCollection(), anyMap());
 414
 415         final MapNode wholeList = ImmutableMapNodeBuilder.create(outerList)
 416                 .withValue(createOuterEntries(listSize, "testing-values")).build();
 417
 418         transaction = producer.createTransaction(false);
 419         cursor = transaction.createCursor(TEST_ID);
 420         assertNotNull(cursor);
 421
 422         cursor.write(wholeList.getIdentifier(), wholeList);
 423         cursor.close();
 424
 425         transaction.submit().checkedGet();
 426
 427         leaderShardFactory.registerListener(mockedDataTreeListener, Collections.singletonList(TEST_ID),
 428                 true, Collections.emptyList());
 429
 430         verify(mockedDataTreeListener, timeout(35000).atLeast(2)).onDataTreeChanged(captorForChanges.capture(),
 431                 captorForSubtrees.capture());
 432         verifyNoMoreInteractions(mockedDataTreeListener);
 433         final List<Map<DOMDataTreeIdentifier, NormalizedNode<?, ?>>> allSubtrees = captorForSubtrees.getAllValues();
 434
 435         final Map<DOMDataTreeIdentifier, NormalizedNode<?, ?>> lastSubtree = allSubtrees.get(allSubtrees.size() - 1);
 436
 437         final NormalizedNode<?, ?> actual = lastSubtree.get(TEST_ID);
 438         assertNotNull(actual);
 439
 440         final NormalizedNode<?, ?> expected =
 441                 ImmutableContainerNodeBuilder.create()
 442                         .withNodeIdentifier(new NodeIdentifier(TestModel.TEST_QNAME))
 443                         .withChild(ImmutableMapNodeBuilder.create(outerList)
 444                                 .withValue(createOuterEntries(listSize, "testing-values")).build())
 445                         .build();
 446
 447
 448         for (final DistributedShardRegistration registration : registrations) {
 449             waitOnAsyncTask(registration.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 450         }
 451
 452         waitOnAsyncTask(testShardReg.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 453
 454         assertEquals(expected, actual);
 455     }
 456
 457     @Test
 458     public void testMultipleRegistrationsAtOnePrefix() throws Exception {
 459         initEmptyDatastores();
 460
 461         for (int i = 0; i < 10; i++) {
 462             LOG.debug("Round {}", i);
 463             final DistributedShardRegistration reg1 = waitOnAsyncTask(leaderShardFactory.createDistributedShard(
 464                     TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
 465                     DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 466
 467             leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 468                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 469
 470             assertNotNull(findLocalShard(leaderDistributedDataStore.getActorContext(),
 471                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
 472
 473             waitOnAsyncTask(reg1.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 474
 475             waitUntilShardIsDown(leaderDistributedDataStore.getActorContext(),
 476                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 477         }
 478     }
 479
 480     @Test
 481     public void testCDSDataTreeProducer() throws Exception {
 482         initEmptyDatastores();
 483
 484         final DistributedShardRegistration reg1 = waitOnAsyncTask(leaderShardFactory.createDistributedShard(
 485                 TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
 486                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 487
 488         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 489                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 490
 491         assertNotNull(findLocalShard(leaderDistributedDataStore.getActorContext(),
 492                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
 493
 494
 495         final DOMDataTreeIdentifier configRoot =
 496                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY);
 497         final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(configRoot));
 498
 499         assertTrue(producer instanceof CDSDataTreeProducer);
 500
 501         final CDSDataTreeProducer cdsProducer = (CDSDataTreeProducer) producer;
 502         CDSShardAccess shardAccess = cdsProducer.getShardAccess(TEST_ID);
 503         assertEquals(shardAccess.getShardIdentifier(), TEST_ID);
 504
 505         shardAccess = cdsProducer.getShardAccess(INNER_LIST_ID);
 506         assertEquals(TEST_ID, shardAccess.getShardIdentifier());
 507
 508         shardAccess = cdsProducer.getShardAccess(configRoot);
 509         assertEquals(configRoot, shardAccess.getShardIdentifier());
 510
 511         waitOnAsyncTask(reg1.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 512     }
 513
 514     private static Collection<MapEntryNode> createOuterEntries(final int amount, final String valuePrefix) {
 515         final Collection<MapEntryNode> ret = new ArrayList<>();
 516         for (int i = 0; i < amount; i++) {
 517             ret.add(ImmutableNodes.mapEntryBuilder()
 518                     .withNodeIdentifier(new NodeIdentifierWithPredicates(TestModel.OUTER_LIST_QNAME,
 519                             QName.create(TestModel.OUTER_LIST_QNAME, "id"), i))
 520                     .withChild(ImmutableNodes
 521                             .leafNode(QName.create(TestModel.OUTER_LIST_QNAME, "id"), i))
 522                     .withChild(createWholeInnerList(amount, "outer id: " + i + " " + valuePrefix))
 523                     .build());
 524         }
 525
 526         return ret;
 527     }
 528
 529     private static MapNode createWholeInnerList(final int amount, final String valuePrefix) {
 530         return ImmutableMapNodeBuilder.create().withNodeIdentifier(new NodeIdentifier(TestModel.INNER_LIST_QNAME))
 531                 .withValue(createInnerListMapEntries(amount, valuePrefix)).build();
 532     }
 533
 534     private static Collection<MapEntryNode> createInnerListMapEntries(final int amount, final String valuePrefix) {
 535         final Collection<MapEntryNode> ret = new ArrayList<>();
 536         for (int i = 0; i < amount; i++) {
 537             ret.add(ImmutableNodes.mapEntryBuilder()
 538                     .withNodeIdentifier(new NodeIdentifierWithPredicates(TestModel.INNER_LIST_QNAME,
 539                             QName.create(TestModel.INNER_LIST_QNAME, "name"), Integer.toString(i)))
 540                     .withChild(ImmutableNodes
 541                             .leafNode(QName.create(TestModel.INNER_LIST_QNAME, "value"), valuePrefix + "-" + i))
 542                     .build());
 543         }
 544
 545         return ret;
 546     }
 547
 548     private static YangInstanceIdentifier getOuterListIdFor(final int id) {
 549         return TestModel.OUTER_LIST_PATH.node(new NodeIdentifierWithPredicates(
 550                 TestModel.OUTER_LIST_QNAME, QName.create(TestModel.OUTER_LIST_QNAME, "id"), id));
 551     }
 552 }