Merge pull request #12 from ps-dev/ADAPT-1802-dataClassification

ADAPT-1802: [hydra-publish] Adapt data classification changes according to our policy

avro/src/main/java/com/pluralsight/hydra/avro/JsonConverter.java

@@ -283,7 +283,10 @@ private Object typeConvert(Object value, String name, Schema schema) throws IOEx
             case STRING:
                 return value.toString();
             case ENUM:
-                boolean valid = schema.getEnumSymbols().contains(value);
+                if (name.equals("dataClassification")) {
+                    value = adaptOldDataClassificationValue(value);
+                }
+                boolean valid = schema.getEnumSymbols().contains(value.toString());
                 if (!valid)
                     throw new IllegalArgumentException(value + " is not a valid symbol. Possible values are: " +
                             schema.getEnumSymbols() + ".");
@@ -335,4 +338,26 @@ private Map cleanAvro(Map<String, Object> oldRaw) {
         }
         return newMap;
     }
+
+    private String adaptOldDataClassificationValue(Object oldValue) {
+        String newValue;
+
+        switch (oldValue.toString()) {
+            case "InternalUseOnly":
+                newValue = "InternalUse";
+                break;
+            case "ConfidentialPII":
+                newValue = "Confidential";
+                break;
+            case "RestrictedFinancial":
+            case "RestrictedEmployeeData":
+                newValue = "Restricted";
+                break;
+            default:
+                newValue = oldValue.toString();
+                break;
+        }
+
+        return newValue;
+    }
 }

core/src/main/scala/hydra/core/marshallers/HydraJsonSupport.scala

@@ -158,7 +158,7 @@ trait HydraJsonSupport extends SprayJsonSupport with DefaultJsonProtocol {
 
   implicit val genericErrorFormat = jsonFormat2(GenericError)
 
-  implicit val topicCreationMetadataFormat = jsonFormat9(TopicMetadataRequest)
+  implicit val topicCreationMetadataFormat = jsonFormat10(TopicMetadataRequest)
 
   implicit val genericSchemaFormat = jsonFormat2(GenericSchema)
 
@@ -172,6 +172,7 @@ case class TopicMetadataRequest(
     derived: Boolean,
     deprecated: Option[Boolean],
     dataClassification: String,
+    subDataClassification: Option[String],
     contact: String,
     additionalDocumentation: Option[String],
     notes: Option[String],

ingest/src/main/scala/hydra.ingest/modules/Bootstrap.scala

@@ -1,7 +1,6 @@
 package hydra.ingest.modules
 
 import java.time.Instant
-
 import cats.data.NonEmptyList
 import cats.effect.Sync
 import cats.syntax.all._
@@ -11,6 +10,7 @@ import hydra.kafka.model._
 import hydra.kafka.programs.CreateTopicProgram
 import hydra.kafka.util.KafkaUtils.TopicDetails
 import hydra.kafka.algebras.{HydraTag, KafkaAdminAlgebra}
+import hydra.kafka.model.DataClassification.InternalUse
 import hydra.kafka.model.TopicMetadataV2Request.Subject
 
 final class Bootstrap[F[_]: MonadError[*[_], Throwable]] private (
@@ -55,7 +55,8 @@ final class Bootstrap[F[_]: MonadError[*[_], Throwable]] private (
             StreamTypeV2.Entity,
             deprecated = false,
             None,
-            InternalUseOnly,
+            InternalUse,
+            Some(SubDataClassification.InternalUseOnly),
             NonEmptyList.of(cfg.contactMethod),
             Instant.now,
             List.empty,
@@ -84,7 +85,8 @@ final class Bootstrap[F[_]: MonadError[*[_], Throwable]] private (
           StreamTypeV2.Entity,
           deprecated = false,
           None,
-          InternalUseOnly,
+          InternalUse,
+          Some(SubDataClassification.InternalUseOnly),
           NonEmptyList.of(dvsConsumersTopicConfig.contactMethod),
           Instant.now,
           List.empty,
@@ -116,7 +118,8 @@ final class Bootstrap[F[_]: MonadError[*[_], Throwable]] private (
           StreamTypeV2.Entity,
           deprecated = false,
           None,
-          InternalUseOnly,
+          InternalUse,
+          Some(SubDataClassification.InternalUseOnly),
           NonEmptyList.of(cooTopicConfig.contactMethod),
           Instant.now,
           List.empty,
@@ -147,7 +150,8 @@ final class Bootstrap[F[_]: MonadError[*[_], Throwable]] private (
         StreamTypeV2.Entity,
         deprecated = false,
         None,
-        InternalUseOnly,
+        InternalUse,
+        Some(SubDataClassification.InternalUseOnly),
         NonEmptyList.of(cooTopicConfig.contactMethod),
         Instant.now,
         List.empty,

ingest/src/test/scala/hydra/ingest/http/SchemasEndpointSpec.scala

@@ -59,6 +59,7 @@ class SchemasEndpointSpec
         false,
         Some(false),
         "Public",
+        Some("Public"),
         "test_contact",
         Some("test_additionalDocumentation"),
         Some("test_notes"),

ingest/src/test/scala/hydra/ingest/programs/TopicDeletionProgramSpec.scala

@@ -15,6 +15,7 @@ import hydra.kafka.algebras.KafkaAdminAlgebra._
 import hydra.kafka.algebras.MetadataAlgebra.TopicMetadataContainer
 import hydra.kafka.algebras.{KafkaAdminAlgebra, KafkaClientAlgebra, TestConsumerGroupsAlgebra, TestMetadataAlgebra}
 import hydra.kafka.model.ContactMethod.Email
+import hydra.kafka.model.DataClassification.Public
 import hydra.kafka.model.TopicConsumer.{TopicConsumerKey, TopicConsumerValue}
 import hydra.kafka.model.TopicMetadataV2Request.Subject
 import hydra.kafka.model._
@@ -138,6 +139,7 @@ class TopicDeletionProgramSpec extends AnyFlatSpec with Matchers {
       deprecated = deprecated,
       deprecatedDate,
       Public,
+      None,
       NonEmptyList.of(Email.create(email).get),
       createdDate,
       List.empty,

ingest/src/test/scala/hydra/ingest/utils/TopicUtils.scala

@@ -6,6 +6,7 @@ import cats.implicits._
 import hydra.kafka.algebras.MetadataAlgebra.TopicMetadataContainer
 import hydra.kafka.algebras.TestMetadataAlgebra
 import hydra.kafka.model.ContactMethod.Email
+import hydra.kafka.model.DataClassification.Public
 import hydra.kafka.model.TopicMetadataV2Request.Subject
 import hydra.kafka.model._
 import org.apache.avro.SchemaBuilder
@@ -25,6 +26,7 @@ object TopicUtils {
         deprecated = false,
         deprecatedDate = None,
         Public,
+        None,
         NonEmptyList.of(Email.create("[email protected]").get),
         Instant.now(),
         List.empty,

ingestors/kafka/src/main/resources/HydraMetadataTopic.avsc

@@ -38,9 +38,28 @@
       "type": {
         "name": "dataClassificationEnum",
         "type": "enum",
-        "symbols": ["Public", "InternalUseOnly", "ConfidentialPII", "RestrictedFinancial","RestrictedEmployeeData"]
+        "symbols": ["Public", "InternalUse", "Confidential", "Restricted"]
       }
     },
+    {
+      "name": "subDataClassification",
+      "type": [
+        "null",
+        {
+          "type": "enum",
+          "name": "SubDataClassification",
+          "namespace": "hydra.kafka.model",
+          "symbols": [
+            "Public",
+            "InternalUseOnly",
+            "ConfidentialPII",
+            "RestrictedFinancial",
+            "RestrictedEmployeeData"
+          ]
+        }
+      ],
+      "default": null
+    },
     {
       "name": "derived",
       "type": "boolean"

ingestors/kafka/src/main/scala/hydra/kafka/model/DataClassification.scala

@@ -0,0 +1,30 @@
+package hydra.kafka.model
+
+import enumeratum.{Enum, EnumEntry}
+
+import scala.collection.immutable
+
+sealed trait DataClassification extends EnumEntry
+
+object DataClassification extends Enum[DataClassification]  {
+
+  case object Public extends DataClassification
+  case object InternalUse extends DataClassification
+  case object Confidential extends DataClassification
+  case object Restricted extends DataClassification
+
+  override val values: immutable.IndexedSeq[DataClassification] = findValues
+}
+
+sealed trait SubDataClassification extends EnumEntry
+
+object SubDataClassification extends Enum[SubDataClassification] {
+
+  case object Public extends SubDataClassification
+  case object InternalUseOnly extends SubDataClassification
+  case object ConfidentialPII extends SubDataClassification
+  case object RestrictedFinancial extends SubDataClassification
+  case object RestrictedEmployeeData extends SubDataClassification
+
+  override val values: immutable.IndexedSeq[SubDataClassification] = findValues
+}

ingestors/kafka/src/main/scala/hydra/kafka/model/TopicMetadata.scala

@@ -2,14 +2,14 @@ package hydra.kafka.model
 
 import java.time.Instant
 import java.util.UUID
-
 import cats.data.Validated.{Invalid, Valid}
 import cats.data.{NonEmptyList, Validated}
 import cats.syntax.all._
 import cats.{Applicative, ApplicativeError, Monad, MonadError}
 import fs2.kafka.Headers
 import hydra.avro.convert.{ISODateConverter, IsoDate}
 import hydra.core.marshallers._
+import hydra.kafka.model.DataClassification._
 import hydra.kafka.model.TopicMetadataV2Request.Subject
 import org.apache.avro.generic.GenericRecord
 import org.apache.avro.io.{Encoder, EncoderFactory}
@@ -31,6 +31,7 @@ case class TopicMetadata(
     derived: Boolean,
     deprecated: Option[Boolean],
     dataClassification: String,
+    subDataClassification: Option[String],
     contact: String,
     additionalDocumentation: Option[String],
     notes: Option[String],
@@ -145,6 +146,7 @@ final case class TopicMetadataV2ValueOptionalTagList(
                                          deprecated: Boolean,
                                          deprecatedDate: Option[Instant],
                                          dataClassification: DataClassification,
+                                         subDataClassification: Option[SubDataClassification],
                                          contact: NonEmptyList[ContactMethod],
                                          createdDate: Instant,
                                          parentSubjects: List[String],
@@ -160,6 +162,7 @@ final case class TopicMetadataV2ValueOptionalTagList(
       deprecated,
       deprecatedDate,
       dataClassification,
+      subDataClassification,
       contact,
       createdDate,
       parentSubjects,
@@ -178,6 +181,7 @@ final case class TopicMetadataV2Value(
     deprecated: Boolean,
     deprecatedDate: Option[Instant],
     dataClassification: DataClassification,
+    subDataClassification: Option[SubDataClassification],
     contact: NonEmptyList[ContactMethod],
     createdDate: Instant,
     parentSubjects: List[String],
@@ -193,6 +197,7 @@ final case class TopicMetadataV2Value(
       deprecated,
       deprecatedDate,
       dataClassification,
+      subDataClassification,
       contact,
       createdDate,
       parentSubjects,
@@ -225,6 +230,29 @@ object TopicMetadataV2ValueOptionalTagList {
 
   implicit val dataClassificationCodec: Codec[DataClassification] =
     Codec.deriveEnum[DataClassification](
+      symbols = List(
+        "Public",
+        "InternalUse",
+        "Confidential",
+        "Restricted"
+      ),
+      encode = {
+        case Public       => "Public"
+        case InternalUse  => "InternalUse"
+        case Confidential => "Confidential"
+        case Restricted   => "Restricted"
+      },
+      decode = {
+        case "Public"       => Right(Public)
+        case "InternalUse"  => Right(InternalUse)
+        case "Confidential" => Right(Confidential)
+        case "Restricted"   => Right(Restricted)
+        case other          => Left(AvroError(s"$other is not a DataClassification. Valid value is one of: ${DataClassification.values}"))
+      }
+    )
+
+  implicit val subDataClassificationCodec: Codec[SubDataClassification] =
+    Codec.deriveEnum[SubDataClassification](
       symbols = List(
         "Public",
         "InternalUseOnly",
@@ -233,19 +261,19 @@ object TopicMetadataV2ValueOptionalTagList {
         "RestrictedEmployeeData"
       ),
       encode = {
-        case Public                 => "Public"
-        case InternalUseOnly        => "InternalUseOnly"
-        case ConfidentialPII        => "ConfidentialPII"
-        case RestrictedFinancial    => "RestrictedFinancial"
-        case RestrictedEmployeeData => "RestrictedEmployeeData"
+        case SubDataClassification.Public                 => "Public"
+        case SubDataClassification.InternalUseOnly        => "InternalUseOnly"
+        case SubDataClassification.ConfidentialPII        => "ConfidentialPII"
+        case SubDataClassification.RestrictedFinancial    => "RestrictedFinancial"
+        case SubDataClassification.RestrictedEmployeeData => "RestrictedEmployeeData"
       },
       decode = {
-        case "Public"                 => Right(Public)
-        case "InternalUseOnly"        => Right(InternalUseOnly)
-        case "ConfidentialPII"        => Right(ConfidentialPII)
-        case "RestrictedFinancial"    => Right(RestrictedFinancial)
-        case "RestrictedEmployeeData" => Right(RestrictedEmployeeData)
-        case other                    => Left(AvroError(s"$other is not a DataClassification"))
+        case "Public"                 => Right(SubDataClassification.Public)
+        case "InternalUseOnly"        => Right(SubDataClassification.InternalUseOnly)
+        case "ConfidentialPII"        => Right(SubDataClassification.ConfidentialPII)
+        case "RestrictedFinancial"    => Right(SubDataClassification.RestrictedFinancial)
+        case "RestrictedEmployeeData" => Right(SubDataClassification.RestrictedEmployeeData)
+        case other                    => Left(AvroError(s"$other is not a SubDataClassification. Valid value is one of: ${SubDataClassification.values}"))
       }
     )
 
@@ -288,6 +316,7 @@ object TopicMetadataV2ValueOptionalTagList {
         field("deprecated", _.deprecated),
         field("deprecatedDate", _.deprecatedDate, default = Some(None)),
         field("dataClassification", _.dataClassification),
+        field("subDataClassification", _.subDataClassification, default = Some(None)),
         field("contact", _.contact),
         field("createdDate", _.createdDate),
         field("parentSubjects", _.parentSubjects),

ingestors/kafka/src/main/scala/hydra/kafka/model/TopicMetadataAdapter.scala

@@ -9,7 +9,7 @@ import spray.json._
   */
 trait TopicMetadataAdapter extends HydraJsonSupport {
 
-  implicit val topicMetadataFormat = jsonFormat12(TopicMetadata)
+  implicit val topicMetadataFormat = jsonFormat13(TopicMetadata)
 
   def streamLink(rel: String, id: String) = rel -> Link(href = s"/streams/$id")