Resolved the issue where the custom database prefix was lost after th…

waggle-dance-api/src/main/java/com/hotels/bdp/waggledance/api/model/AbstractMetaStore.java

@@ -80,6 +80,20 @@ public AbstractMetaStore(
     this.writableDatabaseWhitelist = writableDatabaseWhitelist;
   }
 
+  public AbstractMetaStore(
+          String name,
+          String remoteMetaStoreUris,
+          String databasePrefix,
+          AccessControlType accessControlType,
+          List<String> writableDatabaseWhitelist) {
+    this.name = name;
+    this.remoteMetaStoreUris = remoteMetaStoreUris;
+    this.databasePrefix = databasePrefix;
+    this.accessControlType = accessControlType;
+    this.writableDatabaseWhitelist = writableDatabaseWhitelist;
+  }
+
+
   public static FederatedMetaStore newFederatedInstance(String name, String remoteMetaStoreUris) {
     return new FederatedMetaStore(name, remoteMetaStoreUris);
   }
@@ -95,6 +109,14 @@ public static PrimaryMetaStore newPrimaryInstance(String name, String remoteMeta
     return new PrimaryMetaStore(name, remoteMetaStoreUris, AccessControlType.READ_ONLY);
   }
 
+  public static PrimaryMetaStore newPrimaryInstance(
+          String name,
+          String remoteMetaStoreUris,
+          String databasePrefix,
+          AccessControlType accessControlType) {
+    return new PrimaryMetaStore(name, remoteMetaStoreUris, databasePrefix, accessControlType);
+  }
+
   public String getDatabasePrefix() {
     return databasePrefix;
   }

waggle-dance-api/src/main/java/com/hotels/bdp/waggledance/api/model/PrimaryMetaStore.java

@@ -43,6 +43,24 @@ public PrimaryMetaStore(
     super(name, remoteMetaStoreUris, accessControlType, writableDatabaseWhitelist);
   }
 
+  public PrimaryMetaStore(
+          String name,
+          String remoteMetaStoreUris,
+          String databasePrefix,
+          AccessControlType accessControlType,
+          String... writableDatabaseWhitelist) {
+    this(name, remoteMetaStoreUris, databasePrefix, accessControlType, Arrays.asList(writableDatabaseWhitelist));
+  }
+
+  public PrimaryMetaStore(
+          String name,
+          String remoteMetaStoreUris,
+          String databasePrefix,
+          AccessControlType accessControlType,
+          List<String> writableDatabaseWhitelist) {
+    super(name, remoteMetaStoreUris, databasePrefix, accessControlType, writableDatabaseWhitelist);
+  }
+
   @Override
   public FederationType getFederationType() {
     return FederationType.PRIMARY;

waggle-dance-api/src/test/java/com/hotels/bdp/waggledance/api/model/AbstractMetaStoreTest.java

@@ -154,6 +154,18 @@ public void newPrimaryInstanceWithDefaultAccessControlType() {
     assertThat(primaryMetaStore.getAccessControlType(), is(AccessControlType.READ_ONLY));
   }
 
+  @Test
+  public void newPrimaryPrefixInstance() {
+    AccessControlType access = AccessControlType.READ_AND_WRITE_AND_CREATE;
+    String databasePrefix = "primary_";
+    PrimaryMetaStore primaryMetaStore = AbstractMetaStore.newPrimaryInstance(name, remoteMetaStoreUri, databasePrefix, access);
+    assertThat(primaryMetaStore.getName(), is(name));
+    assertThat(primaryMetaStore.getRemoteMetaStoreUris(), is(remoteMetaStoreUri));
+    assertThat(primaryMetaStore.getDatabasePrefix(), is(databasePrefix));
+    assertThat(primaryMetaStore.getAccessControlType(), is(access));
+  }
+
+
   @Test
   public void mappedDatabases() {
     List<String> mappedDatabases = new ArrayList<>();

waggle-dance-api/src/test/java/com/hotels/bdp/waggledance/api/model/PrimaryMetaStoreTest.java

@@ -36,6 +36,7 @@ public class PrimaryMetaStoreTest extends AbstractMetaStoreTest<PrimaryMetaStore
   private final String remoteMetaStoreUris = "remoteMetaStoreUris";
   private final List<String> whitelist = new ArrayList<>();
   private final AccessControlType accessControlType = AccessControlType.READ_AND_WRITE_ON_DATABASE_WHITELIST;
+  private final String databasePrefix = "primary_";
 
   public PrimaryMetaStoreTest() {
     super(new PrimaryMetaStore());
@@ -120,4 +121,29 @@ public void constructorWithArrayListForWhitelist() {
     assertThat(store.getWritableDatabaseWhiteList(), is(whitelist));
   }
 
+  @Test
+  public void nonEmptyPrefixConstructor() {
+    whitelist.add("databaseOne");
+    whitelist.add("databaseTwo");
+    PrimaryMetaStore store = new PrimaryMetaStore(name, remoteMetaStoreUris, databasePrefix, accessControlType, whitelist.get(0),
+            whitelist.get(1));
+    assertThat(store.getName(), is(name));
+    assertThat(store.getRemoteMetaStoreUris(), is(remoteMetaStoreUris));
+    assertThat(store.getDatabasePrefix(), is(databasePrefix));
+    assertThat(store.getAccessControlType(), is(accessControlType));
+    assertThat(store.getWritableDatabaseWhiteList(), is(whitelist));
+  }
+
+  @Test
+  public void constructorWithPrefixArrayListForWhitelist() {
+    whitelist.add("databaseOne");
+    whitelist.add("databaseTwo");
+    PrimaryMetaStore store = new PrimaryMetaStore(name, remoteMetaStoreUris, databasePrefix, accessControlType, whitelist);
+    assertThat(store.getName(), is(name));
+    assertThat(store.getRemoteMetaStoreUris(), is(remoteMetaStoreUris));
+    assertThat(store.getDatabasePrefix(), is(databasePrefix));
+    assertThat(store.getAccessControlType(), is(accessControlType));
+    assertThat(store.getWritableDatabaseWhiteList(), is(whitelist));
+  }
+
 }

waggle-dance-core/src/main/java/com/hotels/bdp/waggledance/server/security/DatabaseWhitelistAccessControlHandler.java

@@ -74,7 +74,7 @@ public void databaseCreatedNotification(String name) {
     AbstractMetaStore newMetaStore;
     if (metaStore instanceof PrimaryMetaStore) {
       newMetaStore = new PrimaryMetaStore(metaStore.getName(), metaStore.getRemoteMetaStoreUris(),
-          metaStore.getAccessControlType(), newWritableDatabaseWhiteList);
+          metaStore.getDatabasePrefix(), metaStore.getAccessControlType(), newWritableDatabaseWhiteList);
       newMetaStore.setMappedDatabases(mappedDatabases);
     } else {
       throw new WaggleDanceException(

waggle-dance-core/src/main/java/com/hotels/bdp/waggledance/server/security/ReadWriteCreateAccessControlHandler.java

@@ -55,7 +55,7 @@ public void databaseCreatedNotification(String name) {
     AbstractMetaStore newMetaStore;
     if (metaStore instanceof PrimaryMetaStore) {
       newMetaStore = new PrimaryMetaStore(metaStore.getName(), metaStore.getRemoteMetaStoreUris(),
-          metaStore.getAccessControlType());
+          metaStore.getDatabasePrefix(), metaStore.getAccessControlType());
       newMetaStore.setMappedDatabases(mappedDatabases);
     } else {
       throw new WaggleDanceException(

waggle-dance-core/src/test/java/com/hotels/bdp/waggledance/server/security/DatabaseWhitelistAccessControlHandlerTest.java

@@ -49,10 +49,12 @@ public class DatabaseWhitelistAccessControlHandlerTest {
   private @Mock FederationService federationService;
   private @Captor ArgumentCaptor<PrimaryMetaStore> captor;
   private DatabaseWhitelistAccessControlHandler handler;
+  private String databasePrefix = "primary_";
 
   @Before
   public void setUp() {
     when(primaryMetaStore.getWritableDatabaseWhiteList()).thenReturn(whitelist);
+    when(primaryMetaStore.getDatabasePrefix()).thenReturn(databasePrefix);
     handler = new DatabaseWhitelistAccessControlHandler(primaryMetaStore, federationService, true);
   }
 
@@ -84,6 +86,7 @@ public void databaseCreatedNotification() throws Exception {
     PrimaryMetaStore updatedMetastore = captor.getValue();
     assertThat(updatedMetastore.getWritableDatabaseWhiteList().size(), is(3));
     assertThat(updatedMetastore.getWritableDatabaseWhiteList(), contains("writabledb", "userdb.*", "newdb"));
+    assertThat(updatedMetastore.getDatabasePrefix(), is(databasePrefix));
   }
 
   @Test
@@ -93,6 +96,7 @@ public void databaseCreatedNotificationNoDuplicates() throws Exception {
     PrimaryMetaStore updatedMetastore = captor.getValue();
     assertThat(updatedMetastore.getWritableDatabaseWhiteList().size(), is(2));
     assertThat(updatedMetastore.getWritableDatabaseWhiteList(), contains("writabledb", "userdb.*"));
+    assertThat(updatedMetastore.getDatabasePrefix(), is(databasePrefix));
   }
 
   @Test
@@ -104,6 +108,7 @@ public void databaseCreatedNotificationPrimaryNoMapped() {
     PrimaryMetaStore newPrimaryMetaStore = captor.getValue();
     assertThat(newPrimaryMetaStore.getMappedDatabases().size(), is(1));
     assertThat(newPrimaryMetaStore.getMappedDatabases().get(0), is(database));
+    assertThat(newPrimaryMetaStore.getDatabasePrefix(), is(databasePrefix));
   }
 
   @Test
@@ -116,6 +121,7 @@ public void databaseCreatedNotificationPrimaryHasEmptyMapped() {
     PrimaryMetaStore newPrimaryMetaStore = captor.getValue();
     assertThat(newPrimaryMetaStore.getMappedDatabases().size(), is(1));
     assertThat(newPrimaryMetaStore.getMappedDatabases().get(0), is(database));
+    assertThat(newPrimaryMetaStore.getDatabasePrefix(), is(databasePrefix));
   }
 
   @Test
@@ -129,6 +135,7 @@ public void databaseCreatedNotificationPrimaryHasNonEmptyMapped() {
     PrimaryMetaStore newPrimaryMetaStore = captor.getValue();
     assertThat(newPrimaryMetaStore.getMappedDatabases().size(), is(mappedDatabases.size() + 1));
     assertThat(newPrimaryMetaStore.getMappedDatabases().get(mappedDatabases.size()), is(database));
+    assertThat(newPrimaryMetaStore.getDatabasePrefix(), is(databasePrefix));
   }
 
 }

waggle-dance-core/src/test/java/com/hotels/bdp/waggledance/server/security/ReadWriteCreateAccessControlHandlerTest.java

@@ -45,10 +45,12 @@ public class ReadWriteCreateAccessControlHandlerTest {
   private @Mock FederationService federationService;
   private @Captor ArgumentCaptor<PrimaryMetaStore> captor;
   private String database = "database";
+  private String databasePrefix = "primary_";
 
   @Before
   public void setUp() {
     handler = new ReadWriteCreateAccessControlHandler(primaryMetaStore, federationService);
+    when(primaryMetaStore.getDatabasePrefix()).thenReturn(databasePrefix);
   }
 
   @Test
@@ -70,6 +72,7 @@ public void databaseCreatedNotificationPrimaryNoMapped() {
     PrimaryMetaStore newPrimaryMetaStore = captor.getValue();
     assertThat(newPrimaryMetaStore.getMappedDatabases().size(), is(1));
     assertThat(newPrimaryMetaStore.getMappedDatabases().get(0), is(database));
+    assertThat(newPrimaryMetaStore.getDatabasePrefix(), is(databasePrefix));
   }
 
   @Test
@@ -81,6 +84,7 @@ public void databaseCreatedNotificationPrimaryHasEmptyMapped() {
     PrimaryMetaStore newPrimaryMetaStore = captor.getValue();
     assertThat(newPrimaryMetaStore.getMappedDatabases().size(), is(1));
     assertThat(newPrimaryMetaStore.getMappedDatabases().get(0), is(database));
+    assertThat(newPrimaryMetaStore.getDatabasePrefix(), is(databasePrefix));
   }
 
   @Test
@@ -93,6 +97,7 @@ public void databaseCreatedNotificationPrimaryHasNonEmptyMapped() {
     PrimaryMetaStore newPrimaryMetaStore = captor.getValue();
     assertThat(newPrimaryMetaStore.getMappedDatabases().size(), is(mappedDatabases.size() + 1));
     assertThat(newPrimaryMetaStore.getMappedDatabases().get(mappedDatabases.size()), is(database));
+    assertThat(newPrimaryMetaStore.getDatabasePrefix(), is(databasePrefix));
   }
 
 }