Support kafka attributes

Change-Id: I22a3183aeb0ed2d7553c7321e04e1f708813e2ce
Signed-off-by: dglFromAtt <dgl@research.att.com>
Issue-ID: DMAAP-920

6 files changed, 95 insertions, 19 deletions

diff --git a/src/main/java/org/onap/dmaap/dbcapi/model/Topic.java b/src/main/java/org/onap/dmaap/dbcapi/model/Topic.java
index 6364382..c2f278d 100644
--- a/src/main/java/org/onap/dmaap/dbcapi/model/Topic.java
+++ b/src/main/java/org/onap/dmaap/dbcapi/model/Topic.java
@@ -31,6 +31,7 @@ import javax.xml.bind.annotation.XmlRootElement;
 import org.onap.dmaap.dbcapi.util.DmaapConfig;
 
 import org.onap.dmaap.dbcapi.service.DmaapService;
+import org.onap.dmaap.dbcapi.service.TopicService;
 
 
 @XmlRootElement
@@ -45,14 +46,20 @@ public class Topic extends DmaapObject  {
 	private ReplicationType	replicationCase;  
 	private String	globalMrURL;		// optional: URL of global MR to replicate to/from
 	private FqtnType  fqtnStyle;
-	private String	version;
+	private String	version;	
+	private String	partitionCount;
+	private	String	replicationCount;
+
 
 	private	ArrayList<MR_Client> clients;
 
 
 	
 	private static Dmaap dmaap = new DmaapService().getDmaap();
-
+	
+	private static String defaultPartitionCount;
+	private static String defaultReplicationCount;
+	
 	// during unit testing, discovered that presence of dots in some values
 	// creates an unplanned topic namespace as we compose the FQTN.
 	// this may create sensitivity (i.e. 403) for subsequent creation of AAF perms, so best to not allow it 
@@ -118,11 +125,24 @@ public class Topic extends DmaapObject  {
 		//this.dcaeLocationName = dcaeLocationName;
 		this.tnxEnabled = tnxEnabled;
 		this.owner = owner;
+		this.init();
 		this.setLastMod();
+		logger.debug( "Topic constructor w args " + this.getLastMod() );
+	}
+	
+	public Topic init() {
+		DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
+ 		
+		defaultPartitionCount = p.getProperty( "MR.partitionCount", "2");
+		defaultReplicationCount = p.getProperty( "MR.replicationCount", "1");
+		
 		this.setStatus( DmaapObject_Status.NEW );
 		this.replicationCase = ReplicationType.Validator("none");
 		this.fqtnStyle = FqtnType.Validator("none");
-		logger.debug( "Topic constructor " + this.getLastMod() );
+		this.setPartitionCount( defaultPartitionCount );
+		this.setReplicationCount( defaultReplicationCount );
+		
+		return this;
 	}
 
 	// expects a String in JSON format, with known fields to populate Topic object
@@ -144,6 +164,7 @@ public class Topic extends DmaapObject  {
 		this.setStatus( (String) jsonObj.get( "status" ) );
 		this.setReplicationCase( ReplicationType.Validator( (String) jsonObj.get( "replicationCase" ) ));
 		this.setFqtnStyle( FqtnType.Validator( (String) jsonObj.get( "fqtnStyle" ) ) );
+		this.setPartitionCount( (String) jsonObj.get("partitionCount"));
 
 	}
 	public String getFqtn() {
@@ -177,6 +198,18 @@ public class Topic extends DmaapObject  {
 	public void setOwner(String owner) {
 		this.owner = owner;
 	}
+	public String getPartitionCount() {
+		return partitionCount;
+	}
+	public void setPartitionCount(String partitions) {
+		this.partitionCount = partitions;
+	}
+	public String getReplicationCount() {
+		return replicationCount;
+	}
+	public void setReplicationCount(String replicationCount) {
+		this.replicationCount = replicationCount;
+	}
 
 
 	public void setClients(ArrayList<MR_Client> clients) {
@@ -212,13 +245,6 @@ public class Topic extends DmaapObject  {
 		return replicationCase;
 	}
 
-
-
-	/*
-	public void setReplicationCase(String val) {
-		this.replicationCase = ReplicationType.Validator(val);
-	}
-	*/
 	
 	public void setReplicationCase(ReplicationType t) {
 		this.replicationCase = t;
@@ -262,7 +288,12 @@ public class Topic extends DmaapObject  {
 		str.append( this.getFqtn() );
 		str.append( "\", \"topicDescription\": \"");
 		str.append( this.getTopicDescription());
-		str.append( "\", \"partitionCount\": \"2\", \"replicationCount\": \"1\" } ");
+		str.append( "\", \"partitionCount\": \"");
+		str.append( this.getPartitionCount());
+		str.append( "\", \"replicationCount\": \"");
+		str.append( this.getReplicationCount());
+		str.append( "\" } ");
+		
 		logger.info( str.toString() );
 		return str.toString();
 	}
diff --git a/src/main/java/org/onap/dmaap/dbcapi/resources/TopicResource.java b/src/main/java/org/onap/dmaap/dbcapi/resources/TopicResource.java
index 8ed2558..dbeea63 100644
--- a/src/main/java/org/onap/dmaap/dbcapi/resources/TopicResource.java
+++ b/src/main/java/org/onap/dmaap/dbcapi/resources/TopicResource.java
@@ -58,11 +58,16 @@ import org.onap.dmaap.dbcapi.util.DmaapConfig;
 @Authorization
 public class TopicResource extends BaseLoggingClass {
 	private static FqtnType defaultTopicStyle;
+	private static String defaultPartitionCount;
+	private static String defaultReplicationCount;
 	TopicService mr_topicService = new TopicService();
 	
 	public TopicResource() {
 		DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
-		defaultTopicStyle = FqtnType.Validator( p.getProperty("MR.topicStyle", "FQTN_LEGACY_FORMAT"));
+ 		defaultTopicStyle = FqtnType.Validator( p.getProperty("MR.topicStyle", "FQTN_LEGACY_FORMAT"));
+		defaultPartitionCount = p.getProperty( "MR.partitionCount", "2");
+		defaultReplicationCount = p.getProperty( "MR.replicationCount", "1");
+		
 		logger.info( "Setting defaultTopicStyle=" + defaultTopicStyle );
 	}
 		
@@ -119,6 +124,14 @@ public class TopicResource extends BaseLoggingClass {
 			logger.info( "setting defaultTopicStyle=" + defaultTopicStyle + " for topic " + topic.getTopicName() );
 			topic.setFqtnStyle( defaultTopicStyle );
 		}
+		String pc = topic.getPartitionCount();
+		if ( pc == null ) {
+			topic.setPartitionCount(defaultPartitionCount);
+		}
+		String rc = topic.getReplicationCount();
+		if ( rc == null ) {
+			topic.setReplicationCount(defaultReplicationCount);
+		}
 		topic.setLastMod();
 		Boolean flag = false;
 		if (useExisting != null) {
diff --git a/src/main/java/org/onap/dmaap/dbcapi/service/DmaapService.java b/src/main/java/org/onap/dmaap/dbcapi/service/DmaapService.java
index 5aae1d4..5c2c8a3 100644
--- a/src/main/java/org/onap/dmaap/dbcapi/service/DmaapService.java
+++ b/src/main/java/org/onap/dmaap/dbcapi/service/DmaapService.java
@@ -43,6 +43,7 @@ import org.onap.dmaap.dbcapi.logging.DmaapbcLogMessageEnum;
 import org.onap.dmaap.dbcapi.model.ApiError;
 import org.onap.dmaap.dbcapi.model.Dmaap;
 import org.onap.dmaap.dbcapi.model.MR_Client;
+import org.onap.dmaap.dbcapi.model.ReplicationType;
 import org.onap.dmaap.dbcapi.model.Topic;
 import org.onap.dmaap.dbcapi.model.DmaapObject.DmaapObject_Status;
 import org.onap.dmaap.dbcapi.util.DmaapConfig;
@@ -136,6 +137,7 @@ public class DmaapService  extends BaseLoggingClass  {
 		if ( ! dmaap.isStatusValid()  || ! nd.getDmaapName().equals(dmaap.getDmaapName()) || dmaap.getVersion().equals( "0") ) {
 			nd.setLastMod();
 			dmaapholder.update(nd);  //need to set this so the following perms will pick up any new vals.
+			dcaeTopicNs = dmaapholder.get().getTopicNsRoot();
 			ApiPolicy apiPolicy = new ApiPolicy();
 			if ( apiPolicy.getUseAuthClass()) {
 				ApiPerms p = new ApiPerms();
@@ -279,12 +281,14 @@ public class DmaapService  extends BaseLoggingClass  {
 		clients.add( nClient );
 	
 		// initialize Topic
-		Topic mmaTopic = new Topic();
+		Topic mmaTopic = new Topic().init();
 		mmaTopic.setTopicName(dmaap.getBridgeAdminTopic());
 		mmaTopic.setClients(clients);
 		mmaTopic.setOwner("BusController");
 		mmaTopic.setTopicDescription("topic reserved for MirrorMaker Administration");
 		mmaTopic.setTnxEnabled("false");
+		mmaTopic.setPartitionCount("1");  // a single partition should guarantee message order
+		
 		
 		ApiError err = new ApiError();
 		TopicService svc = new TopicService();
diff --git a/src/main/java/org/onap/dmaap/dbcapi/service/TopicService.java b/src/main/java/org/onap/dmaap/dbcapi/service/TopicService.java
index 56ed967..49966d7 100644
--- a/src/main/java/org/onap/dmaap/dbcapi/service/TopicService.java
+++ b/src/main/java/org/onap/dmaap/dbcapi/service/TopicService.java
@@ -251,10 +251,11 @@ public class TopicService extends BaseLoggingClass {
 		String mmAgentRole = p.getProperty("MM.AgentRole");
 		String[] Roles = { mmProvRole, mmAgentRole };
 		String[] actions = { "view", "pub", "sub" };
-		Topic bridgeAdminTopic = new Topic();
+		Topic bridgeAdminTopic = new Topic().init();
 		bridgeAdminTopic.setTopicName( dmaapSvc.getBridgeAdminFqtn() );
 		bridgeAdminTopic.setTopicDescription( "RESERVED topic for MirroMaker Provisioning");
 		bridgeAdminTopic.setOwner( "DBC" );
+		
 		ArrayList<MR_Client> clients = new ArrayList<MR_Client>();
 		for( String role: Roles ) {
 			MR_Client client = new MR_Client();
diff --git a/src/main/resources/schema_10.sql b/src/main/resources/schema_10.sql
new file mode 100644
index 0000000..715af54
--- /dev/null
+++ b/src/main/resources/schema_10.sql
@@ -0,0 +1,32 @@
+---
+-- ============LICENSE_START=======================================================
+-- OpenECOMP - org.onap.dbcapi
+-- ================================================================================
+-- Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+-- ================================================================================
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+-- 
+--      http://www.apache.org/licenses/LICENSE-2.0
+-- 
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+-- ============LICENSE_END=========================================================
+---
+
+
+@alter table topic
+	add column 	partition_count	varchar(10),
+	add column	replication_count varchar(10)
+;
+
+-- set existing topics to have old default values, for display purposes
+update topic
+set partition_count = '2', replication_count = '1';
+
+
+update dmaapbc_sch_ver set version = 10 where version = 9;
diff --git a/src/test/java/org/onap/dmaap/dbcapi/testframework/ReflectionHarness.java b/src/test/java/org/onap/dmaap/dbcapi/testframework/ReflectionHarness.java
index 931ee28..d500fb6 100644
--- a/src/test/java/org/onap/dmaap/dbcapi/testframework/ReflectionHarness.java
+++ b/src/test/java/org/onap/dmaap/dbcapi/testframework/ReflectionHarness.java
@@ -21,15 +21,10 @@ package org.onap.dmaap.dbcapi.testframework;
 
 import static org.junit.Assert.*;
 
-import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Type;
 
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
 import static java.lang.System.out;
 import static java.lang.System.err;