diff --git a/ingest/src/main/scala/hydra.ingest/modules/Bootstrap.scala b/ingest/src/main/scala/hydra.ingest/modules/Bootstrap.scala index d37eb68fa..6825a69f3 100644 --- a/ingest/src/main/scala/hydra.ingest/modules/Bootstrap.scala +++ b/ingest/src/main/scala/hydra.ingest/modules/Bootstrap.scala @@ -55,6 +55,8 @@ final class Bootstrap[F[_]: MonadError[*[_], Throwable]] private ( StreamTypeV2.Entity, deprecated = false, None, + replacementTopics = None, + previousTopics = None, InternalUseOnly, NonEmptyList.of(cfg.contactMethod), Instant.now, @@ -65,6 +67,7 @@ final class Bootstrap[F[_]: MonadError[*[_], Throwable]] private ( Some("Data-Platform"), None, List.empty, + None, None ), TopicDetails(cfg.numPartitions, cfg.replicationFactor, cfg.minInsyncReplicas, Map("cleanup.policy" -> "compact")) @@ -83,6 +86,8 @@ final class Bootstrap[F[_]: MonadError[*[_], Throwable]] private ( StreamTypeV2.Entity, deprecated = false, None, + replacementTopics = None, + previousTopics = None, InternalUseOnly, NonEmptyList.of(dvsConsumersTopicConfig.contactMethod), Instant.now, @@ -93,6 +98,7 @@ final class Bootstrap[F[_]: MonadError[*[_], Throwable]] private ( Some("Data-Platform"), None, List.empty, + None, None ), TopicDetails( @@ -114,6 +120,8 @@ final class Bootstrap[F[_]: MonadError[*[_], Throwable]] private ( StreamTypeV2.Entity, deprecated = false, None, + replacementTopics = None, + previousTopics = None, InternalUseOnly, NonEmptyList.of(cooTopicConfig.contactMethod), Instant.now, @@ -124,6 +132,7 @@ final class Bootstrap[F[_]: MonadError[*[_], Throwable]] private ( Some("Data-Platform"), None, List.empty, + None, None ), TopicDetails( @@ -144,6 +153,8 @@ final class Bootstrap[F[_]: MonadError[*[_], Throwable]] private ( StreamTypeV2.Entity, deprecated = false, None, + replacementTopics = None, + previousTopics = None, InternalUseOnly, NonEmptyList.of(cooTopicConfig.contactMethod), Instant.now, @@ -152,6 +163,7 @@ final class Bootstrap[F[_]: MonadError[*[_], Throwable]] private ( Some("Data-Platform"), None, List.empty, + None, None ), TopicDetails(cfg.numPartitions, cfg.replicationFactor, cfg.minInsyncReplicas, Map("cleanup.policy" -> "compact"))) diff --git a/ingest/src/test/scala/hydra/ingest/programs/TopicDeletionProgramSpec.scala b/ingest/src/test/scala/hydra/ingest/programs/TopicDeletionProgramSpec.scala index 3a8ed888c..0c2d0ef7b 100644 --- a/ingest/src/test/scala/hydra/ingest/programs/TopicDeletionProgramSpec.scala +++ b/ingest/src/test/scala/hydra/ingest/programs/TopicDeletionProgramSpec.scala @@ -137,6 +137,8 @@ class TopicDeletionProgramSpec extends AnyFlatSpec with Matchers { StreamTypeV2.Entity, deprecated = deprecated, deprecatedDate, + replacementTopics = None, + previousTopics = None, Public, NonEmptyList.of(Email.create(email).get), createdDate, @@ -145,7 +147,8 @@ class TopicDeletionProgramSpec extends AnyFlatSpec with Matchers { Some("dvs-teamName"), None, List.empty, - Some("notificationUrl") + Some("notificationUrl"), + None ) private def buildSchema(topic: String, upgrade: Boolean): Schema = { diff --git a/ingest/src/test/scala/hydra/ingest/utils/TopicUtils.scala b/ingest/src/test/scala/hydra/ingest/utils/TopicUtils.scala index c7822ad15..406379f79 100644 --- a/ingest/src/test/scala/hydra/ingest/utils/TopicUtils.scala +++ b/ingest/src/test/scala/hydra/ingest/utils/TopicUtils.scala @@ -26,6 +26,8 @@ object TopicUtils { StreamTypeV2.Entity, deprecated = false, deprecatedDate = None, + replacementTopics = None, + previousTopics = None, Public, NonEmptyList.of(Email.create("test@test.com").get), createdDate, @@ -34,7 +36,8 @@ object TopicUtils { Some("dvs-teamName"), None, List.empty, - Some("notificationUrl") + Some("notificationUrl"), + additionalValidations = None ) val topicMetadataContainer = TopicMetadataContainer( topicMetadataKey, diff --git a/ingestors/kafka/src/main/scala/hydra/kafka/model/AdditionalValidation.scala b/ingestors/kafka/src/main/scala/hydra/kafka/model/AdditionalValidation.scala new file mode 100644 index 000000000..f2205ed0e --- /dev/null +++ b/ingestors/kafka/src/main/scala/hydra/kafka/model/AdditionalValidation.scala @@ -0,0 +1,51 @@ +package hydra.kafka.model + +import enumeratum.{Enum, EnumEntry} +import hydra.kafka.algebras.MetadataAlgebra.TopicMetadataContainer + +import scala.collection.immutable + +sealed trait AdditionalValidation extends EnumEntry + +sealed trait MetadataAdditionalValidation extends AdditionalValidation + +object MetadataAdditionalValidation extends Enum[MetadataAdditionalValidation] { + case object replacementTopics extends MetadataAdditionalValidation + + override val values: immutable.IndexedSeq[MetadataAdditionalValidation] = findValues + + lazy val key: String = "MetadataAdditionalValidation" +} + +object AdditionalValidation { + lazy val allValidations: Option[Map[String, List[AdditionalValidation]]] = + Some(Map( + MetadataAdditionalValidation.key -> MetadataAdditionalValidation.values.toList + )) + + /** + * An OLD topic will have its metadata populated. + * Therefore, additionalValidations=None will be picked from the metadata. + * And no new additionalValidations will be applied on older topics. + * + * A NEW topic will not have a metadata object. + * Therefore, all existing additionalValidations will be assigned. + * Thus, additionalValidations on corresponding fields will be applied. + * + * Corner case: After this feature has been on STAGE/PROD for sometime and some new additionalValidations are required. + * We need not worry about old topics as the value of additionalValidations will remain the same since the topic creation. + * New additionalValidations should be applied only on new topics. + * Therefore, assigning all the values under AdditionalValidation enum is reasonable. + * + * @param metadata a metadata object of current topic + * @return value of additionalValidations if the topic is already existing(OLD topic) otherwise all enum values under AdditionalValidation(NEW topic) + */ + def validations(metadata: Option[TopicMetadataContainer]): Option[Map[String, List[AdditionalValidation]]] = + metadata.map(_.value.additionalValidations).getOrElse(AdditionalValidation.allValidations) + + def metadataValidations(metadata: Option[TopicMetadataContainer]): Option[List[MetadataAdditionalValidation]] = + validations(metadata) flatMap { vMap => + vMap.get(MetadataAdditionalValidation.key) + .map(_.asInstanceOf[List[MetadataAdditionalValidation]]) + } +} diff --git a/ingestors/kafka/src/main/scala/hydra/kafka/model/TopicMetadata.scala b/ingestors/kafka/src/main/scala/hydra/kafka/model/TopicMetadata.scala index 347c4ff21..3cece9b21 100644 --- a/ingestors/kafka/src/main/scala/hydra/kafka/model/TopicMetadata.scala +++ b/ingestors/kafka/src/main/scala/hydra/kafka/model/TopicMetadata.scala @@ -144,6 +144,8 @@ final case class TopicMetadataV2ValueOptionalTagList( streamType: StreamTypeV2, deprecated: Boolean, deprecatedDate: Option[Instant], + replacementTopics: Option[List[String]], + previousTopics: Option[List[String]], dataClassification: DataClassification, contact: NonEmptyList[ContactMethod], createdDate: Instant, @@ -151,13 +153,16 @@ final case class TopicMetadataV2ValueOptionalTagList( notes: Option[String], teamName: Option[String], tags: Option[List[String]], - notificationUrl: Option[String] + notificationUrl: Option[String], + additionalValidations: Option[Map[String, List[AdditionalValidation]]] ) { def toTopicMetadataV2Value: TopicMetadataV2Value = { TopicMetadataV2Value( streamType, deprecated, deprecatedDate, + replacementTopics, + previousTopics, dataClassification, contact, createdDate, @@ -165,7 +170,8 @@ final case class TopicMetadataV2ValueOptionalTagList( notes, teamName, tags.getOrElse(List.empty), - notificationUrl + notificationUrl, + additionalValidations ) } } @@ -175,6 +181,8 @@ final case class TopicMetadataV2Value( streamType: StreamTypeV2, deprecated: Boolean, deprecatedDate: Option[Instant], + replacementTopics: Option[List[String]], + previousTopics: Option[List[String]], dataClassification: DataClassification, contact: NonEmptyList[ContactMethod], createdDate: Instant, @@ -182,13 +190,16 @@ final case class TopicMetadataV2Value( notes: Option[String], teamName: Option[String], tags: List[String], - notificationUrl: Option[String] + notificationUrl: Option[String], + additionalValidations: Option[Map[String, List[AdditionalValidation]]] ) { def toTopicMetadataV2ValueOptionalTagList: TopicMetadataV2ValueOptionalTagList = { TopicMetadataV2ValueOptionalTagList( streamType, deprecated, deprecatedDate, + replacementTopics, + previousTopics, dataClassification, contact, createdDate, @@ -196,7 +207,8 @@ final case class TopicMetadataV2Value( notes, teamName, tags.some, - notificationUrl + notificationUrl, + additionalValidations ) } } @@ -258,6 +270,19 @@ object TopicMetadataV2ValueOptionalTagList { private implicit val contactMethodCodec: Codec[ContactMethod] = Codec.derive[ContactMethod] + private implicit val validationsCodec: Codec[AdditionalValidation] = Codec.deriveEnum[AdditionalValidation]( + symbols = List( + MetadataAdditionalValidation.replacementTopics.entryName + ), + encode = { + case MetadataAdditionalValidation.replacementTopics => MetadataAdditionalValidation.replacementTopics.entryName + }, + decode = { + case "replacementTopics" => Right(MetadataAdditionalValidation.replacementTopics) + case other => Left(AvroError(s"$other is not a ${AdditionalValidation.toString}")) + } + ) + implicit val codec: Codec[TopicMetadataV2ValueOptionalTagList] = Codec.record[TopicMetadataV2ValueOptionalTagList]( name = "TopicMetadataV2Value", @@ -267,6 +292,8 @@ object TopicMetadataV2ValueOptionalTagList { (field("streamType", _.streamType), field("deprecated", _.deprecated), field("deprecatedDate", _.deprecatedDate, default = Some(None)), + field("replacementTopics", _.replacementTopics, default = Some(None)), + field("previousTopics", _.previousTopics, default = Some(None)), field("dataClassification", _.dataClassification), field("contact", _.contact), field("createdDate", _.createdDate), @@ -274,7 +301,8 @@ object TopicMetadataV2ValueOptionalTagList { field("notes", _.notes, default = Some(None)), field("teamName", _.teamName, default = Some(None)), field("tags", _.tags, default = Some(None)), - field("notificationUrl", _.notificationUrl, default = Some(None)) + field("notificationUrl", _.notificationUrl, default = Some(None)), + field("additionalValidations", _.additionalValidations, default = Some(None)) ).mapN(TopicMetadataV2ValueOptionalTagList.apply) } } diff --git a/ingestors/kafka/src/main/scala/hydra/kafka/model/TopicMetadataV2Transport.scala b/ingestors/kafka/src/main/scala/hydra/kafka/model/TopicMetadataV2Transport.scala index df3c9c4e0..f459dbdcf 100644 --- a/ingestors/kafka/src/main/scala/hydra/kafka/model/TopicMetadataV2Transport.scala +++ b/ingestors/kafka/src/main/scala/hydra/kafka/model/TopicMetadataV2Transport.scala @@ -10,9 +10,9 @@ import hydra.kafka.algebras.MetadataAlgebra.TopicMetadataContainer import hydra.kafka.model.TopicMetadataV2Request.Subject import org.apache.avro.Schema import shapeless.Witness -import shapeless.Witness.Lt import java.time.Instant +import scala.collection.immutable sealed trait DataClassification @@ -78,6 +78,8 @@ final case class TopicMetadataV2Request( streamType: StreamTypeV2, deprecated: Boolean, deprecatedDate: Option[Instant], + replacementTopics: Option[List[String]], + previousTopics: Option[List[String]], dataClassification: DataClassification, contact: NonEmptyList[ContactMethod], createdDate: Instant, @@ -86,7 +88,8 @@ final case class TopicMetadataV2Request( teamName: Option[String], numPartitions: Option[TopicMetadataV2Request.NumPartitions], tags: List[String], - notificationUrl: Option[String] + notificationUrl: Option[String], + additionalValidations: Option[Map[String, List[AdditionalValidation]]] ) { def toValue: TopicMetadataV2Value = { @@ -94,6 +97,8 @@ final case class TopicMetadataV2Request( streamType, deprecated, deprecatedDate, + replacementTopics, + previousTopics, dataClassification, contact, createdDate, @@ -101,7 +106,8 @@ final case class TopicMetadataV2Request( notes, teamName, tags, - notificationUrl + notificationUrl, + additionalValidations ) } } @@ -138,6 +144,8 @@ object TopicMetadataV2Request { mor.streamType, mor.deprecated, mor.deprecatedDate, + mor.replacementTopics, + mor.previousTopics, mor.dataClassification, mor.contact, mor.createdDate, @@ -146,7 +154,8 @@ object TopicMetadataV2Request { mor.teamName, mor.numPartitions, mor.tags, - mor.notificationUrl + mor.notificationUrl, + mor.additionalValidations ) } } @@ -160,6 +169,8 @@ final case class TopicMetadataV2Response( streamType: StreamTypeV2, deprecated: Boolean, deprecatedDate: Option[Instant], + replacementTopics: Option[List[String]], + previousTopics: Option[List[String]], dataClassification: DataClassification, contact: NonEmptyList[ContactMethod], createdDate: Instant, @@ -179,6 +190,8 @@ object TopicMetadataV2Response { v.streamType, v.deprecated, v.deprecatedDate, + v.replacementTopics, + v.previousTopics, v.dataClassification, v.contact, v.createdDate, @@ -194,6 +207,8 @@ object TopicMetadataV2Response { final case class MetadataOnlyRequest(streamType: StreamTypeV2, deprecated: Boolean, deprecatedDate: Option[Instant], + replacementTopics: Option[List[String]], + previousTopics: Option[List[String]], dataClassification: DataClassification, contact: NonEmptyList[ContactMethod], createdDate: Instant, @@ -202,7 +217,8 @@ final case class MetadataOnlyRequest(streamType: StreamTypeV2, teamName: Option[String], numPartitions: Option[TopicMetadataV2Request.NumPartitions], tags: List[String], - notificationUrl: Option[String]) { + notificationUrl: Option[String], + additionalValidations: Option[Map[String, List[AdditionalValidation]]]) { } diff --git a/ingestors/kafka/src/main/scala/hydra/kafka/programs/CreateTopicProgram.scala b/ingestors/kafka/src/main/scala/hydra/kafka/programs/CreateTopicProgram.scala index 9a8bc98ce..33e655ee8 100644 --- a/ingestors/kafka/src/main/scala/hydra/kafka/programs/CreateTopicProgram.scala +++ b/ingestors/kafka/src/main/scala/hydra/kafka/programs/CreateTopicProgram.scala @@ -1,19 +1,19 @@ package hydra.kafka.programs -import java.time.Instant import cats.effect.{Bracket, ExitCase, Resource, Sync} +import cats.implicits._ import hydra.avro.registry.SchemaRegistry import hydra.avro.registry.SchemaRegistry.SchemaVersion import hydra.kafka.algebras.{KafkaAdminAlgebra, KafkaClientAlgebra, MetadataAlgebra} -import hydra.kafka.model.{StreamTypeV2, TopicMetadataV2, TopicMetadataV2Key, TopicMetadataV2Request} +import hydra.kafka.model.TopicMetadataV2Request.Subject +import hydra.kafka.model._ import hydra.kafka.programs.CreateTopicProgram._ import hydra.kafka.util.KafkaUtils.TopicDetails -import org.typelevel.log4cats.Logger import org.apache.avro.Schema +import org.typelevel.log4cats.Logger +import retry._ import retry.syntax.all._ -import retry.{RetryDetails, RetryPolicy, _} -import cats.implicits._ -import hydra.kafka.model.TopicMetadataV2Request.Subject +import java.time.Instant import scala.language.higherKinds import scala.util.control.NoStackTrace @@ -24,7 +24,8 @@ final class CreateTopicProgram[F[_]: Bracket[*[_], Throwable]: Sleep: Logger] pr retryPolicy: RetryPolicy[F], v2MetadataTopicName: Subject, metadataAlgebra: MetadataAlgebra[F], - validator: KeyAndValueSchemaV2Validator[F] + schemaValidator: KeyAndValueSchemaV2Validator[F], + metadataValidator: MetadataV2Validator[F] ) (implicit eff: Sync[F]){ private def onFailure(resourceTried: String): (Throwable, RetryDetails) => F[Unit] = { @@ -118,7 +119,10 @@ final class CreateTopicProgram[F[_]: Bracket[*[_], Throwable]: Sleep: Logger] pr None } } - message = (TopicMetadataV2Key(topicName), createTopicRequest.copy(createdDate = createdDate, deprecatedDate = deprecatedDate).toValue) + message = ( + TopicMetadataV2Key(topicName), + createTopicRequest.copy(createdDate = createdDate, deprecatedDate = deprecatedDate, + additionalValidations = AdditionalValidation.validations(metadata)).toValue) records <- TopicMetadataV2.encode[F](message._1, Some(message._2), None) _ <- kafkaClient .publishMessage(records, v2MetadataTopicName.value) @@ -136,7 +140,8 @@ final class CreateTopicProgram[F[_]: Bracket[*[_], Throwable]: Sleep: Logger] pr def createTopicFromMetadataOnly(topicName: Subject, createTopicRequest: TopicMetadataV2Request, withRequiredFields: Boolean = false): F[Unit] = for { _ <- checkThatTopicExists(topicName.value) - _ <- validator.validate(createTopicRequest, topicName, withRequiredFields) + _ <- metadataValidator.validate(createTopicRequest, topicName) + _ <- schemaValidator.validate(createTopicRequest, topicName, withRequiredFields) _ <- publishMetadata(topicName, createTopicRequest) } yield () @@ -158,7 +163,8 @@ final class CreateTopicProgram[F[_]: Bracket[*[_], Throwable]: Sleep: Logger] pr .copy(partialConfig = defaultTopicDetails.configs ++ getCleanupPolicyConfig) (for { - _ <- Resource.eval(validator.validate(createTopicRequest, topicName, withRequiredFields)) + _ <- Resource.eval(metadataValidator.validate(createTopicRequest, topicName)) + _ <- Resource.eval(schemaValidator.validate(createTopicRequest, topicName, withRequiredFields)) _ <- registerSchemas( topicName, createTopicRequest.schemas.key, @@ -187,7 +193,8 @@ object CreateTopicProgram { retryPolicy, v2MetadataTopicName, metadataAlgebra, - KeyAndValueSchemaV2Validator.make(schemaRegistry, metadataAlgebra, defaultLoopHoleCutoffDate) + KeyAndValueSchemaV2Validator.make(schemaRegistry, metadataAlgebra, defaultLoopHoleCutoffDate), + MetadataV2Validator.make(metadataAlgebra) ) } diff --git a/ingestors/kafka/src/main/scala/hydra/kafka/programs/MetadataV2Validator.scala b/ingestors/kafka/src/main/scala/hydra/kafka/programs/MetadataV2Validator.scala new file mode 100644 index 000000000..cba901219 --- /dev/null +++ b/ingestors/kafka/src/main/scala/hydra/kafka/programs/MetadataV2Validator.scala @@ -0,0 +1,69 @@ +package hydra.kafka.programs + +import cats.effect.Sync +import cats.syntax.all._ +import hydra.common.validation.Validator.ValidationChain +import hydra.common.validation.{ValidationError, Validator} +import hydra.kafka.algebras.MetadataAlgebra +import hydra.kafka.model.TopicMetadataV2Request.Subject +import hydra.kafka.model.{AdditionalValidation, MetadataAdditionalValidation, TopicMetadataV2Request} + +import scala.language.higherKinds + +class MetadataV2Validator[F[_] : Sync](metadataAlgebra: MetadataAlgebra[F]) extends Validator { + + def validate(request: TopicMetadataV2Request, subject: Subject): F[Unit] = + for { + metadata <- metadataAlgebra.getMetadataFor(subject) + additionalValidations <- AdditionalValidation.metadataValidations(metadata).getOrElse(List.empty).pure + _ <- resultOf(validate(request)) + _ <- resultOf(validateAdditional(additionalValidations, request, subject)) + } yield() + + private def validate(request: TopicMetadataV2Request): F[List[ValidationChain]] = { + val validationResults = validateTopicsFormat(request.replacementTopics) ++ + validateTopicsFormat(request.previousTopics) + validationResults.pure + } + + private def validateAdditional(additionalValidations: List[MetadataAdditionalValidation], + request: TopicMetadataV2Request, + subject: Subject): F[List[ValidationChain]] = + (additionalValidations flatMap { + case MetadataAdditionalValidation.replacementTopics => + List(validateDeprecatedTopicHasReplacementTopic(request.deprecated, request.replacementTopics, subject.value)) + }).pure + + private def validateDeprecatedTopicHasReplacementTopic(deprecated: Boolean, replacementTopics: Option[List[String]], topic: String): ValidationChain = { + val hasReplacementTopicsIfDeprecated = if (deprecated) replacementTopics.exists(_.nonEmpty) else true + validate(hasReplacementTopicsIfDeprecated, ReplacementTopicsMissingError(topic)) + } + + private def validateTopicsFormat(maybeTopics: Option[List[String]]): List[ValidationChain] = { + val validationResults = for { + topics <- maybeTopics + } yield { + topics map { topic => + val isValidTopicFormat = Subject.createValidated(topic).nonEmpty + validate(isValidTopicFormat, InvalidTopicFormatError(topic)) + } + } + validationResults.getOrElse(List.empty) + } +} + +sealed trait MetadataValidationError extends ValidationError + +case class ReplacementTopicsMissingError(topic: String) extends MetadataValidationError { + override def message: String = s"Field 'replacementTopics' is required when the topic '$topic' is being deprecated!" +} + +case class InvalidTopicFormatError(topic: String) extends MetadataValidationError { + override def message: String = s"$topic : " + Subject.invalidFormat +} + + +object MetadataV2Validator { + def make[F[_] : Sync](metadataAlgebra: MetadataAlgebra[F]): MetadataV2Validator[F] = + new MetadataV2Validator[F](metadataAlgebra) +} \ No newline at end of file diff --git a/ingestors/kafka/src/main/scala/hydra/kafka/serializers/TopicMetadataV2Parser.scala b/ingestors/kafka/src/main/scala/hydra/kafka/serializers/TopicMetadataV2Parser.scala index 5b1a5d8b4..f7bcedb92 100644 --- a/ingestors/kafka/src/main/scala/hydra/kafka/serializers/TopicMetadataV2Parser.scala +++ b/ingestors/kafka/src/main/scala/hydra/kafka/serializers/TopicMetadataV2Parser.scala @@ -1,14 +1,14 @@ package hydra.kafka.serializers import java.time.Instant - import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport import cats.data.Validated.{Invalid, Valid} import cats.data._ import cats.syntax.all._ +import enumeratum.EnumEntry import eu.timepit.refined.auto._ import hydra.kafka.model.ContactMethod.{Email, Slack} -import hydra.kafka.model.TopicMetadataV2Request.Subject +import hydra.kafka.model.TopicMetadataV2Request.{NumPartitions, Subject} import hydra.kafka.model._ import hydra.kafka.serializers.Errors._ import hydra.kafka.serializers.TopicMetadataV2Parser.IntentionallyUnimplemented @@ -270,11 +270,26 @@ sealed trait TopicMetadataV2Parser } + class EnumEntryJsonFormat[E <: EnumEntry](values: Seq[E]) extends RootJsonFormat[E] { + + override def write(obj: E): JsValue = JsString(obj.entryName) + + override def read(json: JsValue): E = json match { + case s: JsString => values.find(v => v.entryName == s.value).getOrElse(deserializationError(s)) + case x => deserializationError(x) + } + + private def deserializationError(value: JsValue) = throw DeserializationException(s"Expected a value from enum $values instead of $value") + } + + implicit val newMetadataValidationFormat: EnumEntryJsonFormat[AdditionalValidation] = + new EnumEntryJsonFormat[AdditionalValidation](Seq.empty) + implicit object TopicMetadataV2Format extends RootJsonFormat[TopicMetadataV2Request] { override def write(obj: TopicMetadataV2Request): JsValue = - jsonFormat13(TopicMetadataV2Request.apply).write(obj) + jsonFormat16(TopicMetadataV2Request.apply).write(obj) override def read(json: JsValue): TopicMetadataV2Request = json match { case j: JsObject => @@ -308,6 +323,59 @@ sealed trait TopicMetadataV2Parser } } + private class Converter(j: JsObject) { + + def toSubject(jsonField: String): Subject = + SubjectFormat.read(j.getFields(jsonField).headOption.getOrElse(JsString.empty)) + + def toStreamTypeV2(jsonField: String): StreamTypeV2 = + StreamTypeV2Format.read( + j.getFields(jsonField) + .headOption + .getOrElse(throwDeserializationError(jsonField, "String"))) + + def toDataClassification(jsonField: String): DataClassification = + DataClassificationFormat.read( + j.getFields(jsonField) + .headOption + .getOrElse( + throwDeserializationError(jsonField, "String"))) + + def toListOfContactMethods(jsonField: String): NonEmptyList[ContactMethod] = + ContactFormat.read( + j.getFields(jsonField) + .headOption + .getOrElse(throwDeserializationError(jsonField, "JsObject"))) + + def toListOfStrings(jsonField: String): List[String] = + j.fields.get(jsonField) match { + case Some(t) => t.convertTo[Option[List[String]]].getOrElse(List.empty) + case None => List.empty[String] + } + + def toOptionalString(jsonField: String): Option[String] = + j.fields.get(jsonField) match { + case Some(teamName) => teamName.convertTo[Option[String]] + case None => throwDeserializationError(jsonField, "String") + } + + def toOptionalStringNoError(jsonField: String): Option[String] = j.getFields(jsonField).headOption.map(_.convertTo[String]) + + def toOptionalNumPartitions(jsonField: String): Option[NumPartitions] = + j.fields.get(jsonField).map { num => + TopicMetadataV2Request.NumPartitions.from(num.convertTo[Int]).toOption match { + case Some(numP) => numP + case None => throwDeserializationError(jsonField, "Int [10-50]") + } + } + + def toOptionalListOfStrings(jsonField: String): Option[List[String]] = + j.fields.get(jsonField) match { + case Some(t) => t.convertTo[Option[List[String]]] + case None => None + } + } + implicit object MetadataOnlyRequestFormat extends RootJsonFormat[MetadataOnlyRequest] { override def write(obj: MetadataOnlyRequest): JsValue = { JsString(obj.toString) @@ -322,18 +390,11 @@ sealed trait TopicMetadataV2Parser def getValidationResult(json: JsValue): MetadataValidationResult[MetadataOnlyRequest] = json match { case j: JsObject => - val subject = toResult( - SubjectFormat - .read(j.getFields("subject").headOption.getOrElse(JsString.empty)) - ) - val streamType = toResult( - StreamTypeV2Format.read( - j.getFields("streamType") - .headOption - .getOrElse(throwDeserializationError("streamType", "String")) - ) - ) - val deprecated = toResult(getBoolWithKey(j, "deprecated")) + val c = new Converter(j) + val subject = toResult(c.toSubject("subject")) + val streamType = toResult(c.toStreamTypeV2("streamType")) + val deprecatedFieldName = "deprecated" + val deprecated = toResult(getBoolWithKey(j, deprecatedFieldName)) val deprecatedDate = if ( deprecated.toOption.getOrElse(false) && !j.getFields("deprecatedDate").headOption.getOrElse(None).equals(None)) { toResult(Option(Instant.parse(j.getFields("deprecatedDate").headOption .getOrElse(throwDeserializationError("deprecatedDate","long")) @@ -341,58 +402,24 @@ sealed trait TopicMetadataV2Parser } else { toResult(None) } - val dataClassification = toResult( - DataClassificationFormat.read( - j.getFields("dataClassification") - .headOption - .getOrElse( - throwDeserializationError("dataClassification", "String") - ) - ) - ) - val contact = toResult( - ContactFormat.read( - j.getFields("contact") - .headOption - .getOrElse(throwDeserializationError("contact", "JsObject")) - ) - ) + val dataClassification = toResult(c.toDataClassification("dataClassification")) + val contact = toResult(c.toListOfContactMethods("contact")) val createdDate = toResult(Instant.now()) - val parentSubjects = toResult( - j.fields.get("parentSubjects") match { - case Some(t) => t.convertTo[Option[List[String]]].getOrElse(List.empty) - case None => List.empty[String] - }) - val notes = toResult( - j.getFields("notes").headOption.map(_.convertTo[String]) - ) - val teamName = toResult( - j.fields.get("teamName") match { - case Some(teamName) => teamName.convertTo[Option[String]] - case None => throwDeserializationError("teamName", "String") - } - ) - val numPartitions = toResult( - j.fields.get("numPartitions").map { num => - TopicMetadataV2Request.NumPartitions.from(num.convertTo[Int]).toOption match { - case Some(numP) => numP - case None => throwDeserializationError("numPartitions", "Int [10-50]") - } - } - ) - val tags = toResult( - j.fields.get("tags") match { - case Some(t) => t.convertTo[Option[List[String]]].getOrElse(List.empty) - case None => List.empty[String] - } - ) - val notificationUrl = toResult( - j.getFields("notificationUrl").headOption.map(_.convertTo[String]) - ) + val parentSubjects = toResult(c.toListOfStrings("parentSubjects")) + val notes = toResult(c.toOptionalStringNoError("notes")) + val teamName = toResult(c.toOptionalString("teamName")) + val numPartitions = toResult(c.toOptionalNumPartitions("numPartitions")) + val tags = toResult(c.toListOfStrings("tags")) + val notificationUrl = toResult(c.toOptionalStringNoError("notificationUrl")) + val replacementTopics = toResult(c.toOptionalListOfStrings("replacementTopics")) + val previousTopics = toResult(c.toOptionalListOfStrings("previousTopics")) + ( streamType, deprecated, deprecatedDate, + replacementTopics, + previousTopics, dataClassification, contact, createdDate, @@ -401,7 +428,8 @@ sealed trait TopicMetadataV2Parser teamName, numPartitions, tags, - notificationUrl + notificationUrl, + toResult(None) // Never pick additionalValidations from the request. ).mapN(MetadataOnlyRequest.apply) } } @@ -422,7 +450,7 @@ sealed trait TopicMetadataV2Parser implicit object TopicMetadataResponseV2Format extends RootJsonFormat[TopicMetadataV2Response] { override def read(json: JsValue): TopicMetadataV2Response = throw IntentionallyUnimplemented - override def write(obj: TopicMetadataV2Response): JsValue = jsonFormat13(TopicMetadataV2Response.apply).write(obj) + override def write(obj: TopicMetadataV2Response): JsValue = jsonFormat15(TopicMetadataV2Response.apply).write(obj) } private def throwDeserializationError(key: String, `type`: String) = @@ -549,5 +577,4 @@ object Errors { final case class MissingField(field: String, fieldType: String) { def errorMessage: String = s"Field `$field` of type $fieldType" } - } diff --git a/ingestors/kafka/src/test/scala/hydra/kafka/algebras/MetadataAlgebraSpec.scala b/ingestors/kafka/src/test/scala/hydra/kafka/algebras/MetadataAlgebraSpec.scala index 4a0d40e55..bee08b106 100644 --- a/ingestors/kafka/src/test/scala/hydra/kafka/algebras/MetadataAlgebraSpec.scala +++ b/ingestors/kafka/src/test/scala/hydra/kafka/algebras/MetadataAlgebraSpec.scala @@ -118,6 +118,8 @@ class MetadataAlgebraSpec extends AnyWordSpecLike with Matchers with Notificatio StreamTypeV2.Entity, deprecated = false, None, + None, + None, Public, NonEmptyList.one(Slack.create("#channel").get), Instant.now, @@ -125,7 +127,8 @@ class MetadataAlgebraSpec extends AnyWordSpecLike with Matchers with Notificatio None, Some("dvs-teamName"), List.empty, - Some("notificationUrl") + Some("notificationUrl"), + None ) (TopicMetadataV2.encode[IO](key, if (nullValue) None else Some(value), None), key, value) } diff --git a/ingestors/kafka/src/test/scala/hydra/kafka/endpoints/BootstrapEndpointV2Spec.scala b/ingestors/kafka/src/test/scala/hydra/kafka/endpoints/BootstrapEndpointV2Spec.scala index f9384fe12..ea51287e0 100644 --- a/ingestors/kafka/src/test/scala/hydra/kafka/endpoints/BootstrapEndpointV2Spec.scala +++ b/ingestors/kafka/src/test/scala/hydra/kafka/endpoints/BootstrapEndpointV2Spec.scala @@ -17,7 +17,7 @@ import hydra.kafka.algebras._ import hydra.kafka.model.ContactMethod.{Email, Slack} import hydra.kafka.model.TopicMetadataV2Request.Subject import hydra.kafka.model._ -import hydra.kafka.programs.CreateTopicProgram +import hydra.kafka.programs.{CreateTopicProgram, InvalidTopicFormatError, ReplacementTopicsMissingError} import hydra.kafka.serializers.TopicMetadataV2Parser._ import hydra.kafka.util.KafkaUtils.TopicDetails import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient @@ -127,21 +127,7 @@ final class BootstrapEndpointV2Spec StreamTypeV2.Entity, deprecated = false, None, - Public, - NonEmptyList.of(Email.create("test@pluralsight.com").get, Slack.create("#dev-data-platform").get), - Instant.now, - List.empty, None, - Some("dvs-teamName"), - None, - List.empty, - Some("notificationUrl") - ).toJson.compactPrint - - val validRequestWithoutDVSTag = TopicMetadataV2Request( - Schemas(getTestSchema("key"), getTestSchema("value")), - StreamTypeV2.Entity, - deprecated = false, None, Public, NonEmptyList.of(Email.create("test@pluralsight.com").get, Slack.create("#dev-data-platform").get), @@ -151,14 +137,17 @@ final class BootstrapEndpointV2Spec Some("dvs-teamName"), None, List.empty, - Some("notificationUrl") + Some("notificationUrl"), + None ).toJson.compactPrint - val validRequestWithDVSTag = TopicMetadataV2Request( + val topicMetadataV2Request = TopicMetadataV2Request( Schemas(getTestSchema("key"), getTestSchema("value")), StreamTypeV2.Entity, deprecated = false, None, + None, + None, Public, NonEmptyList.of(Email.create("test@pluralsight.com").get, Slack.create("#dev-data-platform").get), Instant.now, @@ -166,9 +155,12 @@ final class BootstrapEndpointV2Spec None, Some("dvs-teamName"), None, - List("DVS"), - Some("notificationUrl") - ).toJson.compactPrint + List.empty, + Some("notificationUrl"), + None + ) + + val validRequestWithoutDVSTag = topicMetadataV2Request.toJson.compactPrint "accept a valid request without a DVS tag" in { testCreateTopicProgram @@ -184,6 +176,7 @@ final class BootstrapEndpointV2Spec } "accept a valid request with a DVS tag" in { + val validRequestWithDVSTag = topicMetadataV2Request.copy(tags = List("DVS")).toJson.compactPrint testCreateTopicProgram .map { bootstrapEndpoint => Put("/v2/topics/dvs.testing", HttpEntity(ContentTypes.`application/json`, validRequestWithDVSTag)) ~> Route.seal( @@ -229,6 +222,8 @@ final class BootstrapEndpointV2Spec StreamTypeV2.Entity, deprecated = false, None, + None, + None, Public, NonEmptyList.of(Email.create("test@pluralsight.com").get, Slack.create("#dev-data-platform").get), Instant.now, @@ -237,7 +232,8 @@ final class BootstrapEndpointV2Spec None, None, List.empty, - Some("notificationUrl") + Some("notificationUrl"), + None ).toJson.compactPrint testCreateTopicProgram .map { bootstrapEndpoint => @@ -302,6 +298,8 @@ final class BootstrapEndpointV2Spec StreamTypeV2.Entity, deprecated = false, None, + None, + None, Public, NonEmptyList.of(Email.create("test@pluralsight.com").get, Slack.create("#dev-data-platform").get), Instant.now, @@ -310,7 +308,8 @@ final class BootstrapEndpointV2Spec Some("dvs-teamName"), None, List("DVS"), - Some("notificationUrl") + Some("notificationUrl"), + None ).toJson.compactPrint @@ -330,6 +329,8 @@ final class BootstrapEndpointV2Spec StreamTypeV2.Entity, deprecated = false, None, + None, + None, Public, NonEmptyList.of(Email.create("test@pluralsight.com").get, Slack.create("#dev-data-platform").get), Instant.now, @@ -338,7 +339,8 @@ final class BootstrapEndpointV2Spec Some("dvs-teamName"), None, List("Source: NotValid"), - Some("notificationUrl") + Some("notificationUrl"), + None ).toJson.compactPrint implicit val notificationSenderMock: InternalNotificationSender[IO] = getInternalNotificationSenderMock[IO] @@ -354,5 +356,73 @@ final class BootstrapEndpointV2Spec } }.unsafeRunSync() } + + "reject a request when a topic being deprecated does not have replacementTopics" in { + val deprecateWithoutReplacementTopicsRequest = topicMetadataV2Request.copy(deprecated = true).toJson.compactPrint + + testFailure(deprecateWithoutReplacementTopicsRequest, error = ReplacementTopicsMissingError("dvs.testing").message) + } + + "reject a request when a topic being deprecated contains replacementTopics with invalid patterns" in { + val deprecateWithInvalidReplacementTopicsRequest = topicMetadataV2Request.copy( + deprecated = true, + replacementTopics = Some(List("dvs.testing", "invalid.dvs.testing")) + ).toJson.compactPrint + + testFailure(deprecateWithInvalidReplacementTopicsRequest, error = InvalidTopicFormatError("invalid.dvs.testing").message) + } + + "reject a request when a topic contains previousTopics with invalid patterns" in { + val deprecateWithInvalidPreviousTopicsRequest = topicMetadataV2Request.copy( + previousTopics = Some(List("dvs.testing", "invalid.dvs.testing")) + ).toJson.compactPrint + + testFailure(deprecateWithInvalidPreviousTopicsRequest, error = InvalidTopicFormatError("invalid.dvs.testing").message) + } + + "accept a request when a topic being deprecated has valid replacementTopics" in { + val deprecateWithValidReplacementTopicsRequest = topicMetadataV2Request.copy( + deprecated = true, + replacementTopics = Some(List("dvs.testing.replacement")) + ).toJson.compactPrint + + testSuccess(deprecateWithValidReplacementTopicsRequest) + } + + "create topic with valid replacementTopics" in { + val replacementTopicsRequest = topicMetadataV2Request.copy( + previousTopics = Some(List("dvs.testing.replacement")) + ).toJson.compactPrint + + testSuccess(replacementTopicsRequest) + } + + "create topic with valid previousTopics" in { + val previousTopicsRequest = topicMetadataV2Request.copy( + previousTopics = Some(List("dvs.testing.previous")) + ).toJson.compactPrint + + testSuccess(previousTopicsRequest) + } + + def testFailure(request: String, error: String) = testRequest(request, reject = true, errorMessage = Some(error)) + + def testSuccess(request: String) = testRequest(request) + + def testRequest(request: String, reject: Boolean = false, errorMessage: Option[String] = None) = + testCreateTopicProgram + .map { bootstrapEndpoint => + Put("/v2/topics/dvs.testing", HttpEntity(ContentTypes.`application/json`, request)) ~> Route.seal( + bootstrapEndpoint.route + ) ~> check { + if (reject) { + response.status shouldBe StatusCodes.BadRequest + errorMessage.foreach(e => responseAs[String] shouldBe e) + } else { + response.status shouldBe StatusCodes.OK + } + } + } + .unsafeRunSync() } } diff --git a/ingestors/kafka/src/test/scala/hydra/kafka/endpoints/TopicMetadataEndpointSpec.scala b/ingestors/kafka/src/test/scala/hydra/kafka/endpoints/TopicMetadataEndpointSpec.scala index 8d77aca68..d4fc93f77 100644 --- a/ingestors/kafka/src/test/scala/hydra/kafka/endpoints/TopicMetadataEndpointSpec.scala +++ b/ingestors/kafka/src/test/scala/hydra/kafka/endpoints/TopicMetadataEndpointSpec.scala @@ -227,7 +227,7 @@ class TopicMetadataEndpointSpec "sends back an error response if topic already exists" in { implicit val timeout = RouteTestTimeout(5.seconds) - EmbeddedKafka.createCustomTopic("testExisting")(kafkaConfig) + createCustomTopic("testExisting")(kafkaConfig) val config = Map( "min.insync.replicas" -> "1", "cleanup.policy" -> "compact", @@ -284,6 +284,7 @@ class TopicMetadataEndpointSpec val validRequest = """{ | "streamType": "Event", | "deprecated": true, + | "replacementTopics": ["dvs.test.subject.new"], | "dataClassification": "InternalUseOnly", | "contact": { | "email": "bob@myemail.com" @@ -314,6 +315,7 @@ class TopicMetadataEndpointSpec """{ | "streamType": "Event", | "deprecated": true, + | "replacementTopics": ["dvs.test.subject.new"], | "dataClassification": "InternalUseOnly", | "contact": { | "email": "bob@myemail.com" diff --git a/ingestors/kafka/src/test/scala/hydra/kafka/model/TopicMetadataSpec.scala b/ingestors/kafka/src/test/scala/hydra/kafka/model/TopicMetadataSpec.scala index 52cb03558..60d1eb8c8 100644 --- a/ingestors/kafka/src/test/scala/hydra/kafka/model/TopicMetadataSpec.scala +++ b/ingestors/kafka/src/test/scala/hydra/kafka/model/TopicMetadataSpec.scala @@ -48,6 +48,8 @@ final class TopicMetadataSpec extends AnyFlatSpecLike with Matchers { StreamTypeV2.Entity, false, None, + None, + None, Public, NonEmptyList.of(ContactMethod.create("test@test.com").get), createdDate, @@ -55,6 +57,7 @@ final class TopicMetadataSpec extends AnyFlatSpecLike with Matchers { None, Some("dvs-teamName"), List.empty, + None, None ) @@ -73,6 +76,8 @@ final class TopicMetadataSpec extends AnyFlatSpecLike with Matchers { |"streamType":"Entity", |"deprecated": false, |"deprecatedDate": null, + |"replacementTopics": null, + |"previousTopics": null, |"dataClassification":"Public", |"teamName":{"string":"dvs-teamName"}, |"contact":[ @@ -84,7 +89,8 @@ final class TopicMetadataSpec extends AnyFlatSpecLike with Matchers { |"parentSubjects": [], |"notes": null, |"notificationUrl": null, - |"tags": null + |"tags": null, + |"additionalValidations": null |}""".stripMargin val decoder = DecoderFactory.get().jsonDecoder(valueSchema, json) @@ -101,6 +107,8 @@ final class TopicMetadataSpec extends AnyFlatSpecLike with Matchers { StreamTypeV2.Entity, false, None, + None, + None, Public, NonEmptyList.of(ContactMethod.create("test@test.com").get), createdDate, @@ -108,7 +116,8 @@ final class TopicMetadataSpec extends AnyFlatSpecLike with Matchers { None, Some("dvs-teamName"), List.empty, - Some("notificationUrl") + Some("notificationUrl"), + None ) val (encodedKey, encodedValue, headers) = diff --git a/ingestors/kafka/src/test/scala/hydra/kafka/programs/CreateTopicProgramSpec.scala b/ingestors/kafka/src/test/scala/hydra/kafka/programs/CreateTopicProgramSpec.scala index a4107a513..3ad19232b 100644 --- a/ingestors/kafka/src/test/scala/hydra/kafka/programs/CreateTopicProgramSpec.scala +++ b/ingestors/kafka/src/test/scala/hydra/kafka/programs/CreateTopicProgramSpec.scala @@ -213,7 +213,7 @@ class CreateTopicProgramSpec extends AsyncFreeSpec with Matchers with IOSuite { "ingest metadata into the metadata topic" in { for { publishTo <- Ref[IO].of(Map.empty[String, (GenericRecord, Option[GenericRecord], Option[Headers])]) - topicMetadata <- TopicMetadataV2.encode[IO](topicMetadataKey, Some(topicMetadataValue)) + topicMetadata <- TopicMetadataV2.encode[IO](topicMetadataKey, Some(topicMetadataValue.copy(additionalValidations = AdditionalValidation.allValidations))) ts <- initTestServices(new TestKafkaClientAlgebraWithPublishTo(publishTo).some) _ <- ts.program.createTopic(subject, topicMetadataRequest, topicDetails, true) published <- publishTo.get @@ -227,7 +227,7 @@ class CreateTopicProgramSpec extends AsyncFreeSpec with Matchers with IOSuite { publishTo <- Ref[IO].of(Map.empty[String, (GenericRecord, Option[GenericRecord], Option[Headers])]) consumeFrom <- Ref[IO].of(Map.empty[Subject, TopicMetadataContainer]) metadata <- IO(new TestMetadataAlgebraWithPublishTo(consumeFrom)) - m <- TopicMetadataV2.encode[IO](topicMetadataKey, Some(topicMetadataValue)) + m <- TopicMetadataV2.encode[IO](topicMetadataKey, Some(topicMetadataValue.copy(additionalValidations = AdditionalValidation.allValidations))) updatedM <- TopicMetadataV2.encode[IO](topicMetadataKey, Some(updatedValue.copy(createdDate = topicMetadataValue.createdDate))) ts <- initTestServices(new TestKafkaClientAlgebraWithPublishTo(publishTo).some, metadata.some) _ <- ts.program.createTopic(subject, topicMetadataRequest, TopicDetails(1, 1, 1), true) @@ -264,6 +264,7 @@ class CreateTopicProgramSpec extends AsyncFreeSpec with Matchers with IOSuite { "ingest updated metadata into the metadata topic - verify deprecated date if supplied is not overwritten" in { val request = createTopicMetadataRequest(keySchema, valueSchema, deprecated = true, deprecatedDate = Some(Instant.now)) + .copy(replacementTopics = Some(List("dvs.subject.replacement"))) val updatedRequest = createTopicMetadataRequest(keySchema, valueSchema, "updated@email.com", deprecated = true) for { publishTo <- Ref[IO].of(Map.empty[String, (GenericRecord, Option[GenericRecord], Option[Headers])]) @@ -2077,6 +2078,228 @@ class CreateTopicProgramSpec extends AsyncFreeSpec with Matchers with IOSuite { result.attempt.map(_ shouldBe UnsupportedLogicalType(valueSchema.getField("timestamp"), "iso-datetime").asLeft) } + + "additionalValidations field is NOT populated if an existing topic does not have it" in { + for { + publishTo <- Ref[IO].of(Map.empty[String, (GenericRecord, Option[GenericRecord], Option[Headers])]) + consumeFrom <- Ref[IO].of(Map.empty[Subject, TopicMetadataContainer]) + metadata <- IO(new TestMetadataAlgebraWithPublishTo(consumeFrom)) + _ <- metadata.addToMetadata(subject, topicMetadataRequest) + ts <- initTestServices(new TestKafkaClientAlgebraWithPublishTo(publishTo).some, metadata.some) + _ <- ts.program.createTopic(subject, topicMetadataRequest, topicDetails, withRequiredFields = true) + published <- publishTo.get + expectedTopicMetadata <- TopicMetadataV2.encode[IO](topicMetadataKey, Some(topicMetadataValue)) // additionalValidations empty in topicMetadataValue + } yield { + published shouldBe Map(metadataTopic -> (expectedTopicMetadata._1, expectedTopicMetadata._2, None)) + } + } + + "additionalValidations field is NOT populated via the additionalValidations field in the create topic request" in { + val requestWithEmptyValidations = createTopicMetadataRequest(keySchema, valueSchema, additionalValidations = AdditionalValidation.allValidations) + + for { + publishTo <- Ref[IO].of(Map.empty[String, (GenericRecord, Option[GenericRecord], Option[Headers])]) + consumeFrom <- Ref[IO].of(Map.empty[Subject, TopicMetadataContainer]) + metadata <- IO(new TestMetadataAlgebraWithPublishTo(consumeFrom)) + _ <- metadata.addToMetadata(subject, topicMetadataRequest) + ts <- initTestServices(new TestKafkaClientAlgebraWithPublishTo(publishTo).some, metadata.some) + _ <- ts.program.createTopic(subject, requestWithEmptyValidations, topicDetails, withRequiredFields = true) + published <- publishTo.get + expectedTopicMetadata <- TopicMetadataV2.encode[IO](topicMetadataKey, Some(topicMetadataValue)) // additionalValidations empty in topicMetadataValue + } yield { + published shouldBe Map(metadataTopic -> (expectedTopicMetadata._1, expectedTopicMetadata._2, None)) + } + } + + "additionalValidations field is populated for a new topic" in { + for { + publishTo <- Ref[IO].of(Map.empty[String, (GenericRecord, Option[GenericRecord], Option[Headers])]) + consumeFrom <- Ref[IO].of(Map.empty[Subject, TopicMetadataContainer]) + metadata <- IO(new TestMetadataAlgebraWithPublishTo(consumeFrom)) + ts <- initTestServices(new TestKafkaClientAlgebraWithPublishTo(publishTo).some, metadata.some) + _ <- ts.program.createTopic(subject, topicMetadataRequest, topicDetails, withRequiredFields = true) + published <- publishTo.get + expectedTopicMetadata <- TopicMetadataV2.encode[IO]( + topicMetadataKey, + Some(topicMetadataValue.copy(additionalValidations = AdditionalValidation.allValidations))) // additionalValidations populated in topicMetadataValue + } yield { + published shouldBe Map(metadataTopic -> (expectedTopicMetadata._1, expectedTopicMetadata._2, None)) + } + } + + "additionalValidations field will remain populated if an existing topic already has it" in { + for { + publishTo <- Ref[IO].of(Map.empty[String, (GenericRecord, Option[GenericRecord], Option[Headers])]) + consumeFrom <- Ref[IO].of(Map.empty[Subject, TopicMetadataContainer]) + metadata <- IO(new TestMetadataAlgebraWithPublishTo(consumeFrom)) + ts <- initTestServices(new TestKafkaClientAlgebraWithPublishTo(publishTo).some, metadata.some) + _ <- ts.program.createTopic(subject, topicMetadataRequest, topicDetails, withRequiredFields = true) + publishedFirst <- publishTo.get + _ <- ts.program.createTopic(subject, topicMetadataRequest, topicDetails, withRequiredFields = true) + publishedSecond <- publishTo.get + expectedTopicMetadata <- TopicMetadataV2.encode[IO] ( + topicMetadataKey, + Some(topicMetadataValue.copy(additionalValidations = AdditionalValidation.allValidations))) // additionalValidations populated in topicMetadataValue + + } yield { + publishedFirst shouldBe Map(metadataTopic -> (expectedTopicMetadata._1, expectedTopicMetadata._2, None)) + publishedSecond shouldBe Map(metadataTopic -> (expectedTopicMetadata._1, expectedTopicMetadata._2, None)) + } + } + + "[existing-topic] When additionalValidations is empty no corresponding validation is done" in { + val deprecateWithoutReplacementTopicsRequest = topicMetadataRequest.copy(deprecated = true) + + val result = for { + publishTo <- Ref[IO].of(Map.empty[String, (GenericRecord, Option[GenericRecord], Option[Headers])]) + consumeFrom <- Ref[IO].of(Map.empty[Subject, TopicMetadataContainer]) + metadata <- IO(new TestMetadataAlgebraWithPublishTo(consumeFrom)) + _ <- metadata.addToMetadata(subject, topicMetadataRequest) + ts <- initTestServices(new TestKafkaClientAlgebraWithPublishTo(publishTo).some, metadata.some) + _ <- ts.program.createTopic(subject, deprecateWithoutReplacementTopicsRequest, topicDetails, withRequiredFields = true) + } yield () + + result.attempt.map(_ shouldBe Right()) + } + + "[new-topic] When additionalValidations is populated corresponding additional validations are done" in { + val deprecateWithoutReplacementTopicsRequest = topicMetadataRequest.copy(deprecated = true) + + val result = for { + publishTo <- Ref[IO].of(Map.empty[String, (GenericRecord, Option[GenericRecord], Option[Headers])]) + consumeFrom <- Ref[IO].of(Map.empty[Subject, TopicMetadataContainer]) + metadata <- IO(new TestMetadataAlgebraWithPublishTo(consumeFrom)) + ts <- initTestServices(new TestKafkaClientAlgebraWithPublishTo(publishTo).some, metadata.some) + _ <- ts.program.createTopic(subject, deprecateWithoutReplacementTopicsRequest, topicDetails, withRequiredFields = true) + } yield () + + result.attempt.map(_ shouldBe ReplacementTopicsMissingError(subject.value).asLeft) + } + + "throw error when one of topics pattern in replacementTopics is incorrect" in { + val incorrectReplacementTopicsRequest = topicMetadataRequest.copy(replacementTopics = Some(List("dvs.valid.replacement", "incorrect.dvs.replacement"))) + val result = for { + ts <- initTestServices() + _ <- ts.program.createTopic(subject, incorrectReplacementTopicsRequest, topicDetails) + } yield () + + result.attempt.map(_ shouldBe InvalidTopicFormatError("incorrect.dvs.replacement").asLeft) + } + + "throw error when the more than one topic patterns in replacementTopics are incorrect" in { + val incorrectReplacementTopicsRequest = topicMetadataRequest.copy( + replacementTopics = Some(List("dvs.valid.replacement", "incorrect.dvs.replacement1", "incorrect.dvs.replacement2"))) + val result = for { + ts <- initTestServices() + _ <- ts.program.createTopic(subject, incorrectReplacementTopicsRequest, topicDetails) + } yield () + + result.attempt.map(_ shouldBe + ValidationCombinedErrors(List( + InvalidTopicFormatError("incorrect.dvs.replacement1").message, + InvalidTopicFormatError("incorrect.dvs.replacement2").message, + )).asLeft) + } + + "throw error when a topic pattern in previousTopics is incorrect" in { + val incorrectPreviousTopicsRequest = topicMetadataRequest.copy(previousTopics = Some(List("dvs.valid.replacement", "incorrect.dvs.previous"))) + val result = for { + ts <- initTestServices() + _ <- ts.program.createTopic(subject, incorrectPreviousTopicsRequest, topicDetails) + } yield () + + result.attempt.map(_ shouldBe InvalidTopicFormatError("incorrect.dvs.previous").asLeft) + } + + "throw error when the more than one topic pattern in previousTopics are incorrect" in { + val incorrectPreviousTopicsRequest = topicMetadataRequest.copy( + previousTopics = Some(List("dvs.valid.previous", "incorrect.dvs.previous1", "incorrect.dvs.previous2"))) + val result = for { + ts <- initTestServices() + _ <- ts.program.createTopic(subject, incorrectPreviousTopicsRequest, topicDetails) + } yield () + + result.attempt.map(_ shouldBe + ValidationCombinedErrors(List( + InvalidTopicFormatError("incorrect.dvs.previous1").message, + InvalidTopicFormatError("incorrect.dvs.previous2").message, + )).asLeft) + } + + "throw error when a topic being deprecated does not have replacementTopics populated" in { + val incorrectTopicDeprecationRequest = topicMetadataRequest.copy(deprecated = true) + + testFailure(incorrectTopicDeprecationRequest, ReplacementTopicsMissingError(subject.value)) + } + + "throw error when a topic being deprecated has empty replacementTopics" in { + val incorrectTopicDeprecationRequest = topicMetadataRequest.copy(deprecated = true, replacementTopics = Some(List.empty)) + + testFailure(incorrectTopicDeprecationRequest, ReplacementTopicsMissingError(subject.value)) + } + + "Topic is deprecated with valid replacementTopics" in { + val topics = Some(List("dvs.subject.replacement")) + val now = Some(Instant.now()) + val deprecateWithReplacementTopicsRequest = topicMetadataRequest.copy( + deprecated = true, + deprecatedDate = now, + replacementTopics = topics) + + testSuccess(deprecateWithReplacementTopicsRequest, + deprecated = true, + deprecatedDate = now, + replacementTopics = topics) + } + + "valid replacementTopics value is accepted and updated" in { + val topics = Some(List("dvs.subject.replacement")) + val replacementTopicsRequest = topicMetadataRequest.copy(replacementTopics = topics) + + testSuccess(replacementTopicsRequest, replacementTopics = topics) + } + + "valid previousTopics value is accepted and updated" in { + val topics = Some(List("dvs.subject.previous")) + val previousTopicsRequest = topicMetadataRequest.copy(previousTopics = topics) + + testSuccess(previousTopicsRequest, previousTopics = topics) + } + + def testSuccess(request: TopicMetadataV2Request, + deprecated: Boolean = false, + deprecatedDate: Option[Instant] = None, + replacementTopics: Option[List[String]] = None, + previousTopics: Option[List[String]] = None) = { + for { + publishTo <- Ref[IO].of(Map.empty[String, (GenericRecord, Option[GenericRecord], Option[Headers])]) + consumeFrom <- Ref[IO].of(Map.empty[Subject, TopicMetadataContainer]) + metadata <- IO(new TestMetadataAlgebraWithPublishTo(consumeFrom)) + ts <- initTestServices(new TestKafkaClientAlgebraWithPublishTo(publishTo).some, metadata.some) + _ <- ts.program.createTopic(subject, request, topicDetails) + published <- publishTo.get + expectedTopicMetadata <- TopicMetadataV2.encode[IO]( + topicMetadataKey, + Some(topicMetadataValue.copy( + deprecated = deprecated, + deprecatedDate = deprecatedDate, + replacementTopics = replacementTopics, + previousTopics = previousTopics, + additionalValidations = AdditionalValidation.allValidations + ))) + } yield { + published shouldBe Map(metadataTopic -> (expectedTopicMetadata._1, expectedTopicMetadata._2, None)) + } + } + + def testFailure(request: TopicMetadataV2Request, error: MetadataValidationError) = { + val result = for { + ts <- initTestServices() + _ <- ts.program.createTopic(subject, request, topicDetails) + } yield () + + result.attempt.map(_ shouldBe error.asLeft) + } def createTopic(createdAtDefaultValue: Option[Long], updatedAtDefaultValue: Option[Long])(implicit createdDate: Instant) = for { @@ -2197,13 +2420,16 @@ object CreateTopicProgramSpec extends NotificationsTestSuite { createdDate: Instant = Instant.now(), deprecated: Boolean = false, deprecatedDate: Option[Instant] = None, - numPartitions: Option[NumPartitions] = None + numPartitions: Option[NumPartitions] = None, + additionalValidations: Option[Map[String, List[AdditionalValidation]]] = None ): TopicMetadataV2Request = TopicMetadataV2Request( Schemas(keySchema, valueSchema), StreamTypeV2.Entity, deprecated = deprecated, deprecatedDate, + replacementTopics = None, + previousTopics = None, Public, NonEmptyList.of(Email.create(email).get), createdDate, @@ -2212,7 +2438,8 @@ object CreateTopicProgramSpec extends NotificationsTestSuite { Some("dvs-teamName"), numPartitions, List.empty, - Some("notification.url") + Some("notification.url"), + additionalValidations = additionalValidations ) def createEventStreamTypeTopicMetadataRequest( @@ -2230,6 +2457,8 @@ object CreateTopicProgramSpec extends NotificationsTestSuite { StreamTypeV2.Event, deprecated = deprecated, deprecatedDate, + replacementTopics = None, + previousTopics = None, Public, NonEmptyList.of(Email.create(email).get), createdDate, @@ -2238,7 +2467,8 @@ object CreateTopicProgramSpec extends NotificationsTestSuite { Some("dvs-teamName"), numPartitions, tags, - Some("notification.url") + Some("notification.url"), + None ) def getSchema(name: String, diff --git a/ingestors/kafka/src/test/scala/hydra/kafka/serializers/TopicMetadataV2ParserSpec.scala b/ingestors/kafka/src/test/scala/hydra/kafka/serializers/TopicMetadataV2ParserSpec.scala index 8dff317cd..7074b9f74 100644 --- a/ingestors/kafka/src/test/scala/hydra/kafka/serializers/TopicMetadataV2ParserSpec.scala +++ b/ingestors/kafka/src/test/scala/hydra/kafka/serializers/TopicMetadataV2ParserSpec.scala @@ -350,6 +350,8 @@ class TopicMetadataV2ParserSpec extends AnyWordSpecLike with Matchers { streamType, deprecated, None, + None, + None, dataClassification, NonEmptyList(email, slackChannel :: Nil), tmv2.createdDate, @@ -358,6 +360,7 @@ class TopicMetadataV2ParserSpec extends AnyWordSpecLike with Matchers { Some(teamName), None, List.empty, + None, None ) } @@ -394,6 +397,8 @@ class TopicMetadataV2ParserSpec extends AnyWordSpecLike with Matchers { streamType, deprecated = false, None, + replacementTopics = None, + previousTopics = None, dataClassification, NonEmptyList(email, slackChannel :: Nil), tmv2.createdDate, @@ -402,6 +407,7 @@ class TopicMetadataV2ParserSpec extends AnyWordSpecLike with Matchers { Some(teamName), None, List.empty, + None, None ) } @@ -588,6 +594,8 @@ class TopicMetadataV2ParserSpec extends AnyWordSpecLike with Matchers { streamType = streamType, deprecated = deprecated, deprecatedDate, + replacementTopics = None, + previousTopics = None, dataClassification = dataClassification, contact = contact, createdDate = createdDate, @@ -596,7 +604,8 @@ class TopicMetadataV2ParserSpec extends AnyWordSpecLike with Matchers { teamName = Some(teamName), numPartitions = np, tags = tags, - notificationUrl = notificationUrl + notificationUrl = notificationUrl, + additionalValidations = None ) TopicMetadataV2Format.write(topicMetadataV2) shouldBe createJsValueOfTopicMetadataV2Request( @@ -625,15 +634,15 @@ class TopicMetadataV2ParserSpec extends AnyWordSpecLike with Matchers { "TopicMetadataV2Format write matches TopicMetadataResponseV2Format write" in { val subject = Subject.createValidated("dvs.valid").get val tmc = TopicMetadataContainer(TopicMetadataV2Key(subject), - TopicMetadataV2Value(StreamTypeV2.Entity, false, None, Public, + TopicMetadataV2Value(StreamTypeV2.Entity, false, None, None, None, Public, NonEmptyList.one(ContactMethod.create("blah@pluralsight.com").get), - Instant.now(), List.empty, None, Some("dvs-teamName"), List.empty, None), + Instant.now(), List.empty, None, Some("dvs-teamName"), List.empty, None, None), Some(new SchemaFormat(isKey = true).read(validAvroSchema)), Some(new SchemaFormat(isKey = false).read(validAvroSchema))) val response = TopicMetadataV2Response.fromTopicMetadataContainer(tmc) val request = TopicMetadataV2Request.apply(Schemas(tmc.keySchema.get, tmc.valueSchema.get),tmc.value.streamType, - tmc.value.deprecated,tmc.value.deprecatedDate,tmc.value.dataClassification,tmc.value.contact, - tmc.value.createdDate,tmc.value.parentSubjects,tmc.value.notes, teamName = tmc.value.teamName, None, List.empty, None) + tmc.value.deprecated,tmc.value.deprecatedDate,tmc.value.replacementTopics,tmc.value.previousTopics,tmc.value.dataClassification,tmc.value.contact, + tmc.value.createdDate,tmc.value.parentSubjects,tmc.value.notes, teamName = tmc.value.teamName, None, List.empty, None, None) TopicMetadataV2Format.write(request).compactPrint shouldBe TopicMetadataResponseV2Format.write(response).compactPrint.replace(",\"subject\":\"dvs.valid\"", "") @@ -669,50 +678,60 @@ class TopicMetadataV2ParserSpec extends AnyWordSpecLike with Matchers { } "make sure deprecatedDate works with deprecated true None for Deprecated Date" in { - val subject = Subject.createValidated("dvs.valid").get - val before = Instant.now - val tmc = TopicMetadataContainer(TopicMetadataV2Key(subject), - TopicMetadataV2Value(StreamTypeV2.Entity, true, None, - Public, NonEmptyList.one(ContactMethod.create("blah@pluralsight.com").get), Instant.now(), List.empty, None, Some("dvs-teamName"), List.empty, None), - Some(new SchemaFormat(isKey = true).read(validAvroSchema)), - Some(new SchemaFormat(isKey = false).read(validAvroSchema))) - val request = TopicMetadataV2Request.apply(Schemas(tmc.keySchema.get, tmc.valueSchema.get),tmc.value.streamType, - tmc.value.deprecated,tmc.value.deprecatedDate,tmc.value.dataClassification,tmc.value.contact, - tmc.value.createdDate,tmc.value.parentSubjects,tmc.value.notes,tmc.value.teamName, None, List.empty, None) - val firstDeprecatedDate = TopicMetadataV2Format.read(request.toJson).deprecatedDate.getOrElse(None) + val firstDeprecatedDate = topicMetadataV2Request(deprecated = true, replacementTopics = Some(List("dvs.valid.new"))).deprecatedDate firstDeprecatedDate shouldBe None } "make sure deprecatedDate works with deprecated true Instant for Deprecated Date" in { - val subject = Subject.createValidated("dvs.valid").get val now = Instant.now - val tmc = TopicMetadataContainer(TopicMetadataV2Key(subject), - TopicMetadataV2Value(StreamTypeV2.Entity, true, Some(now), - Public, NonEmptyList.one(ContactMethod.create("blah@pluralsight.com").get), Instant.now(), List.empty, None, Some("dvs-teamName"), List.empty, None), - Some(new SchemaFormat(isKey = true).read(validAvroSchema)), - Some(new SchemaFormat(isKey = false).read(validAvroSchema))) - val request = TopicMetadataV2Request.apply(Schemas(tmc.keySchema.get, tmc.valueSchema.get),tmc.value.streamType, - tmc.value.deprecated,tmc.value.deprecatedDate,tmc.value.dataClassification,tmc.value.contact,tmc.value.createdDate, - tmc.value.parentSubjects,tmc.value.notes,tmc.value.teamName, None, List.empty, None) - val firstDeprecatedDate = TopicMetadataV2Format.read(request.toJson).deprecatedDate.get + val firstDeprecatedDate = topicMetadataV2Request(deprecated = true, deprecatedDate = Some(now), + replacementTopics = Some(List("dvs.valid.new")) + ).deprecatedDate.get val now2 = Instant.now now2.isAfter(firstDeprecatedDate) shouldBe true now shouldBe firstDeprecatedDate } "make sure deprecatedDate works with deprecated false" in { - val subject = Subject.createValidated("dvs.valid").get - val tmc = TopicMetadataContainer(TopicMetadataV2Key(subject), - TopicMetadataV2Value(StreamTypeV2.Entity, false, None, - Public, NonEmptyList.one(ContactMethod.create("blah@pluralsight.com").get), Instant.now(), List.empty, None, Some("dvs-teamName"), List.empty, None), - Some(new SchemaFormat(isKey = true).read(validAvroSchema)), - Some(new SchemaFormat(isKey = false).read(validAvroSchema))) - val request = TopicMetadataV2Request.apply(Schemas(tmc.keySchema.get, tmc.valueSchema.get),tmc.value.streamType, - tmc.value.deprecated,tmc.value.deprecatedDate,tmc.value.dataClassification,tmc.value.contact, - tmc.value.createdDate,tmc.value.parentSubjects,tmc.value.notes, tmc.value.teamName, None, List.empty, None) - val firstDeprecatedDate = TopicMetadataV2Format.read(request.toJson).deprecatedDate.getOrElse(None) + val firstDeprecatedDate = topicMetadataV2Request().deprecatedDate firstDeprecatedDate shouldBe None } + "replacementTopics field is populated only when provided" in { + val emptyReplacementTopics = topicMetadataV2Request().replacementTopics + emptyReplacementTopics shouldBe None + + val replacementTopics = Some(List("dvs.valid.replacement")) + val populatedReplacementTopics = topicMetadataV2Request(replacementTopics = replacementTopics).replacementTopics + populatedReplacementTopics shouldBe replacementTopics + } + + "previousTopics field is populated only when provided" in { + val emptyPreviousTopics = topicMetadataV2Request().previousTopics + emptyPreviousTopics shouldBe None + + val previousTopics = Some(List("dvs.valid.previous")) + val populatedPreviousTopics = topicMetadataV2Request(previousTopics = previousTopics).previousTopics + populatedPreviousTopics shouldBe previousTopics + } + } + + private def topicMetadataV2Request( + deprecated: Boolean = false, + deprecatedDate: Option[Instant] = None, + replacementTopics: Option[List[String]] = None, + previousTopics: Option[List[String]] = None, + subject: Subject = Subject.createValidated("dvs.valid").get + ): TopicMetadataV2Request = { + val tmc = TopicMetadataContainer(TopicMetadataV2Key(subject), + TopicMetadataV2Value(StreamTypeV2.Entity, deprecated, deprecatedDate, replacementTopics, previousTopics, + Public, NonEmptyList.one(ContactMethod.create("blah@pluralsight.com").get), Instant.now(), List.empty, None, Some("dvs-teamName"), List.empty, None, None), + Some(new SchemaFormat(isKey = true).read(validAvroSchema)), + Some(new SchemaFormat(isKey = false).read(validAvroSchema))) + val request = TopicMetadataV2Request.apply(Schemas(tmc.keySchema.get, tmc.valueSchema.get), tmc.value.streamType, + tmc.value.deprecated, tmc.value.deprecatedDate, tmc.value.replacementTopics, tmc.value.previousTopics, tmc.value.dataClassification, tmc.value.contact, + tmc.value.createdDate, tmc.value.parentSubjects, tmc.value.notes, tmc.value.teamName, None, List.empty, None, None) + + TopicMetadataV2Format.read(request.toJson) } } diff --git a/ingestors/kafka/src/test/scala/hydra/kafka/utils/TopicUtils.scala b/ingestors/kafka/src/test/scala/hydra/kafka/utils/TopicUtils.scala index 3e88716ec..bed403711 100644 --- a/ingestors/kafka/src/test/scala/hydra/kafka/utils/TopicUtils.scala +++ b/ingestors/kafka/src/test/scala/hydra/kafka/utils/TopicUtils.scala @@ -3,14 +3,12 @@ package hydra.kafka.utils import cats.data.NonEmptyList import cats.effect.IO import cats.implicits._ -import hydra.avro.registry.SchemaRegistry -import hydra.avro.registry.SchemaRegistry.SchemaId import hydra.kafka.algebras.MetadataAlgebra.TopicMetadataContainer import hydra.kafka.algebras.TestMetadataAlgebra import hydra.kafka.model.ContactMethod.Email import hydra.kafka.model.TopicMetadataV2Request.Subject import hydra.kafka.model._ -import org.apache.avro.{Schema, SchemaBuilder} +import org.apache.avro.SchemaBuilder import java.time.Instant @@ -26,6 +24,8 @@ object TopicUtils { StreamTypeV2.Entity, deprecated = false, deprecatedDate = None, + replacementTopics = None, + previousTopics = None, Public, NonEmptyList.of(Email.create("test@test.com").get), createdDate, @@ -34,7 +34,8 @@ object TopicUtils { Some("dvs-teamName"), None, List.empty, - Some("notificationUrl") + Some("notificationUrl"), + additionalValidations = None ) val topicMetadataContainer = TopicMetadataContainer( topicMetadataKey, diff --git a/project/Dependencies.scala b/project/Dependencies.scala index 5f1cc714c..79cd6c04a 100644 --- a/project/Dependencies.scala +++ b/project/Dependencies.scala @@ -34,6 +34,7 @@ object Dependencies { val scalaTestEmbeddedRedisVersion = "0.4.0" val scalaChillBijectionVersion = "0.10.0" val awsSdkVersion = "2.17.192" + val enumeratumVersion = "1.7.2" object Compile { @@ -152,6 +153,7 @@ object Dependencies { "com.fasterxml.jackson.core" % "jackson-databind" % jacksonDatabindVersion ) + val enumeratum = "com.beachape" %% "enumeratum" % enumeratumVersion } // oneOf test, it, main @@ -200,7 +202,7 @@ object Dependencies { val integrationDeps: Seq[ModuleID] = testContainers ++ TestLibraries.getTestLibraries(module = "it") val baseDeps: Seq[ModuleID] = - akka ++ Seq(avro, ciris, refined) ++ cats ++ logging ++ joda ++ testDeps ++ kafkaClients ++ awsMskIamAuth + akka ++ Seq(avro, ciris, refined, enumeratum) ++ cats ++ logging ++ joda ++ testDeps ++ kafkaClients ++ awsMskIamAuth val avroDeps: Seq[ModuleID] = baseDeps ++ confluent ++ jackson ++ guavacache ++ catsEffect ++ redisCache