diff --git a/hw-kafka-avro.cabal b/hw-kafka-avro.cabal --- a/hw-kafka-avro.cabal +++ b/hw-kafka-avro.cabal @@ -1,7 +1,7 @@ cabal-version: 2.4 name: hw-kafka-avro -version: 6.0.2 +version: 6.1.2 synopsis: Avro support for Kafka infrastructure description: Avro support for Kafka infrastructure. category: Services diff --git a/src/Kafka/Avro/Encode.hs b/src/Kafka/Avro/Encode.hs --- a/src/Kafka/Avro/Encode.hs +++ b/src/Kafka/Avro/Encode.hs @@ -36,9 +36,7 @@ valueSubject (Subject subj) = Subject (subj <> "-value") {-# INLINE valueSubject #-} --- | Encodes a provided value as a message key. --- --- Registers the schema in SchemaRegistry with "-key" subject. +-- | Encodes a provided value as a message key with "-key" subject. encodeKey :: (MonadIO m, HasAvroSchema a, ToAvro a) => SchemaRegistry -> Subject @@ -47,9 +45,7 @@ encodeKey sr subj = encode sr (keySubject subj) {-# INLINE encodeKey #-} --- | Encodes a provided value as a message key. --- --- Registers the schema in SchemaRegistry with "-key" subject. +-- | Encodes a provided value as a message key with "-key" subject. encodeKeyWithSchema :: (MonadIO m, ToAvro a) => SchemaRegistry -> Subject @@ -59,9 +55,7 @@ encodeKeyWithSchema sr subj = encodeWithSchema sr (keySubject subj) {-# INLINE encodeKeyWithSchema #-} --- | Encodes a provided value as a message value. --- --- Registers the schema in SchemaRegistry with "-value" subject. +-- | Encodes a provided value as a message value with "-value" subject. encodeValue :: (MonadIO m, HasAvroSchema a, ToAvro a) => SchemaRegistry -> Subject @@ -70,9 +64,7 @@ encodeValue sr subj = encode sr (valueSubject subj) {-# INLINE encodeValue #-} --- | Encodes a provided value as a message value. --- --- Registers the schema in SchemaRegistry with "-value" subject. +-- | Encodes a provided value as a message value with "-value" subject. encodeValueWithSchema :: (MonadIO m, ToAvro a) => SchemaRegistry -> Subject @@ -91,7 +83,6 @@ {-# INLINE encode #-} -- | Encodes a provided value into Avro --- and registers value's schema in SchemaRegistry. encodeWithSchema :: forall a m. (MonadIO m, ToAvro a) => SchemaRegistry -> Subject diff --git a/src/Kafka/Avro/SchemaRegistry.hs b/src/Kafka/Avro/SchemaRegistry.hs --- a/src/Kafka/Avro/SchemaRegistry.hs +++ b/src/Kafka/Avro/SchemaRegistry.hs @@ -3,6 +3,7 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TupleSections #-} module Kafka.Avro.SchemaRegistry ( schemaRegistry, loadSchema, sendSchema @@ -12,42 +13,49 @@ , getGlobalConfig, getSubjectConfig , getVersions, isCompatible , getSubjects +, defaultSchemaRegistryConfig +, cfgAuth +, cfgHeaders +, cfgAutoRegisterSchemas , SchemaId(..), Subject(..) +, SchemaRegistryConfig , SchemaRegistry, SchemaRegistryError(..) , Schema(..) , Compatibility(..), Version(..) ) where -import Control.Arrow (first) -import Control.Exception (throwIO, SomeException (SomeException)) -import Control.Exception.Safe (try, MonadCatch) -import Control.Lens (view, (&), (.~), (^.), (%~)) -import Control.Monad (void) -import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Arrow (first) +import Control.Exception (SomeException (SomeException), throwIO) +import Control.Exception.Safe (MonadCatch, try) +import Control.Lens (view, (%~), (&), (.~), (^.)) +import Control.Monad (void) +import Control.Monad.Except (liftEither) +import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.Trans.Except (ExceptT (ExceptT), except, runExceptT, withExceptT) import Data.Aeson -import Data.Aeson.Types (typeMismatch) -import qualified Data.Aeson.Key as A -import qualified Data.Aeson.KeyMap as KM -import Data.Avro.Schema.Schema (Schema (..), typeName) -import Data.Foldable (traverse_) -import Data.Functor (($>)) -import Data.Bifunctor (bimap) -import Data.Cache as C -import Data.Hashable (Hashable) -import qualified Data.HashMap.Lazy as HM -import Data.Int (Int32) -import Data.String (IsString) -import Data.Text (Text, append, cons, unpack) -import qualified Data.Text.Encoding as Text -import qualified Data.Text.Lazy.Encoding as LText -import Data.Word (Word32) -import GHC.Exception (SomeException, displayException, fromException) -import GHC.Generics (Generic) -import Network.HTTP.Client (HttpException (..), HttpExceptionContent (..), Manager, defaultManagerSettings, newManager, responseStatus) -import Network.HTTP.Types.Header (Header) -import Network.HTTP.Types.Status (notFound404) -import qualified Network.Wreq as Wreq +import qualified Data.Aeson.Key as A +import qualified Data.Aeson.KeyMap as KM +import Data.Aeson.Types (typeMismatch) +import Data.Avro.Schema.Schema (Schema (..), typeName) +import Data.Bifunctor (bimap) +import Data.Cache as C +import Data.Foldable (traverse_) +import Data.Functor (($>)) +import Data.Hashable (Hashable) +import qualified Data.HashMap.Lazy as HM +import Data.Int (Int32) +import Data.List (find) +import Data.String (IsString) +import Data.Text (Text, append, cons, unpack) +import qualified Data.Text.Encoding as Text +import qualified Data.Text.Lazy.Encoding as LText +import Data.Word (Word32) +import GHC.Exception (SomeException, displayException, fromException) +import GHC.Generics (Generic) +import Network.HTTP.Client (HttpException (..), HttpExceptionContent (..), Manager, defaultManagerSettings, newManager, responseStatus) +import Network.HTTP.Types.Header (Header) +import Network.HTTP.Types.Status (notFound404) +import qualified Network.Wreq as Wreq newtype SchemaId = SchemaId { unSchemaId :: Int32} deriving (Eq, Ord, Show, Hashable) newtype SchemaName = SchemaName Text deriving (Eq, Ord, IsString, Show, Hashable) @@ -64,22 +72,36 @@ | BackwardCompatibility deriving (Eq, Show, Ord) +data SchemaRegistryConfig = SchemaRegistryConfig + { cAuth :: Maybe Wreq.Auth + , cExtraHeaders :: [Header] + , cAutoRegisterSchemas :: Bool + } + data SchemaRegistry = SchemaRegistry { srCache :: Cache SchemaId Schema , srReverseCache :: Cache (Subject, SchemaName) SchemaId , srBaseUrl :: String - , srAuth :: Maybe Wreq.Auth - , srExtraHeaders :: [Header] + , srConfig :: SchemaRegistryConfig } data SchemaRegistryError = SchemaRegistryConnectError String | SchemaRegistryLoadError SchemaId | SchemaRegistrySchemaNotFound SchemaId | SchemaRegistrySubjectNotFound Subject + | SchemaRegistryNoCompatibleSchemaFound Schema | SchemaRegistryUrlNotFound String | SchemaRegistrySendError String + | SchemaRegistryCacheError deriving (Show, Eq) +defaultSchemaRegistryConfig :: SchemaRegistryConfig +defaultSchemaRegistryConfig = SchemaRegistryConfig + { cAuth = Nothing + , cExtraHeaders = [] + , cAutoRegisterSchemas = True + } + schemaRegistry :: MonadIO m => String -> m SchemaRegistry schemaRegistry = schemaRegistry_ Nothing @@ -87,14 +109,32 @@ schemaRegistry_ auth = schemaRegistryWithHeaders auth [] schemaRegistryWithHeaders :: MonadIO m => Maybe Wreq.Auth -> [Header] -> String -> m SchemaRegistry -schemaRegistryWithHeaders auth headers url = liftIO $ +schemaRegistryWithHeaders auth headers url + = schemaRegistryWithConfig url $ cfgAuth auth $ cfgHeaders headers defaultSchemaRegistryConfig + +schemaRegistryWithConfig :: MonadIO m => String -> SchemaRegistryConfig -> m SchemaRegistry +schemaRegistryWithConfig url config = liftIO $ SchemaRegistry <$> newCache Nothing <*> newCache Nothing <*> pure url - <*> pure auth - <*> pure headers + <*> pure config +-- | Add authentication options +cfgAuth :: Maybe Wreq.Auth -> SchemaRegistryConfig -> SchemaRegistryConfig +cfgAuth auth config = config { cAuth = auth } + +-- | Add extra headers +cfgHeaders :: [Header] -> SchemaRegistryConfig -> SchemaRegistryConfig +cfgHeaders headers config = config { cExtraHeaders = headers } + +-- | Set whether to auto-publish schemas +-- If set to 'False', encoding will fail if there is no compatible schema +-- in the schema registy. +-- This is equivalent to the confluent 'auto.register.schemas' option. +cfgAutoRegisterSchemas :: Bool -> SchemaRegistryConfig -> SchemaRegistryConfig +cfgAutoRegisterSchemas autoRegisterSchemas config = config { cAutoRegisterSchemas = autoRegisterSchemas } + loadSchema :: MonadIO m => SchemaRegistry -> SchemaId -> m (Either SchemaRegistryError Schema) loadSchema sr sid = do sc <- cachedSchema sr sid @@ -117,7 +157,7 @@ schemaId <- getData "id" wrapped case (,) <$> schema <*> schemaId of - Left err -> return $ Left err + Left err -> return $ Left err Right (RegisteredSchema schema, schemaId) -> cacheSchema sr schemaId schema $> Right schema where @@ -135,19 +175,46 @@ Error e -> Left e +-- | Get the schema ID. +-- If the 'SchemaRegistry' is configured to auto-register schemas, +-- this posts the schema to the schema registry server. +-- Otherwise, this searches for a compatible schema and returns a 'SchemaRegistryNoCompatibleSchemaFound' +-- if none is found. sendSchema :: MonadIO m => SchemaRegistry -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId) sendSchema sr subj sc = do + let schemaName = fullTypeName sc sid <- cachedId sr subj schemaName case sid of Just sid' -> return (Right sid') - Nothing -> do - res <- liftIO $ putSchema sr subj (RegisteredSchema sc) - traverse_ (cacheId sr subj schemaName) res - traverse_ (\sid' -> cacheSchema sr sid' sc) res - pure res - where - schemaName = fullTypeName sc + Nothing -> if cAutoRegisterSchemas (srConfig sr) + then registerSchema sr subj sc + else getCompatibleSchema sr subj sc +registerSchema :: MonadIO m => SchemaRegistry -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId) +registerSchema sr subj sc = do + let schemaName = fullTypeName sc + res <- liftIO $ putSchema sr subj (RegisteredSchema sc) + traverse_ (cacheId sr subj schemaName) res + traverse_ (\sid' -> cacheSchema sr sid' sc) res + pure res + +getCompatibleSchema :: MonadIO m => SchemaRegistry -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId) +getCompatibleSchema sr subj sc = liftIO . runExceptT $ do + let schemaName = fullTypeName sc + versions <- liftEither =<< getVersions sr subj + compatibilites <- liftEither . sequenceA + =<< traverse (\ver -> fmap (,ver) <$> isCompatible sr subj ver sc) versions + let mCompatibleVersion = snd <$> find fst compatibilites + compatibleVersion <- liftEither + $ case mCompatibleVersion of + Just version -> Right version + Nothing -> Left $ SchemaRegistryNoCompatibleSchemaFound sc + _ <- liftEither =<< loadSubjectSchema sr subj compatibleVersion -- caches the schema ID + mSid <- cachedId sr subj schemaName + liftEither $ case mSid of + Just sid' -> pure sid' + Nothing -> Left SchemaRegistryCacheError + getVersions :: MonadIO m => SchemaRegistry -> Subject -> m (Either SchemaRegistryError [Version]) getVersions sr subj@(Subject sbj) = liftIO . runExceptT $ do let url = srBaseUrl sr ++ "/subjects/" ++ unpack sbj ++ "/versions" @@ -180,7 +247,7 @@ let url = srBaseUrl sr ++ "/config" respE <- liftIO . try $ Wreq.getWith (wreqOpts sr) url pure $ case respE of - Left exc -> Left $ wrapError exc + Left exc -> Left $ wrapError exc Right resp -> bimap wrapError (view Wreq.responseBody) (Wreq.asJSON resp) getSubjectConfig :: MonadIO m => SchemaRegistry -> Subject -> m (Either SchemaRegistryError Compatibility) @@ -202,8 +269,8 @@ let accept = ["application/vnd.schemaregistry.v1+json", "application/vnd.schemaregistry+json", "application/json"] acceptHeader = Wreq.header "Accept" .~ accept - authHeader = Wreq.auth .~ srAuth sr - extraHeaders = Wreq.headers %~ (++ srExtraHeaders sr) + authHeader = Wreq.auth .~ cAuth (srConfig sr) + extraHeaders = Wreq.headers %~ (++ cExtraHeaders (srConfig sr)) in Wreq.defaults & acceptHeader & authHeader & extraHeaders getSchemaById :: SchemaRegistry -> SchemaId -> IO (Either SchemaRegistryError RegisteredSchema)