AWS.Lambda.Events.Kafka

It is possible to subscribe Lambda functions to Kafka topics. You can subscribe to topics from Amazon Managed Streaming for Kafka (MSK) as well as self-managed Kafka clusters.

Lambda considers Amazon Managed Streaming for Kafka (MSK) to be a different event source type from a self-managed Apache Kafka cluster, but their payloads are very similar. The types in this module are derived from inspecting invocation payloads, and from reading the following links:

• https://docs.aws.amazon.com/lambda/latest/dg/with-kafka.html
• https://docs.aws.amazon.com/lambda/latest/dg/with-msk.html
• https://aws.amazon.com/blogs/compute/using-amazon-msk-as-an-event-source-for-aws-lambda/
• https://aws.amazon.com/blogs/compute/using-self-hosted-apache-kafka-as-an-event-source-for-aws-lambda/

Documentation

data KafkaEvent

Represents an event from either Amazon MSK or a self-managed Apache Kafka cluster, as the payloads are very similar.

The ToJSON and FromJSON instances on Record perform base64 conversion for you.

See the AWS documentation for a sample payload.

Constructors

KafkaEvent

Fields

• eventSource :: !EventSource
• eventSourceArn :: !(Maybe Text)

  Only present when using the "Amazon MSK" event source mapping.

• bootstrapServers :: !(NonEmpty Text)
• records :: !(Map Text [Record])

  The map's keys look like "${topicName}-${partitionNumber}"

Instances

Eq KafkaEvent

Defined in AWS.Lambda.Events.Kafka

Methods

(==) :: KafkaEvent -> KafkaEvent -> Bool

(/=) :: KafkaEvent -> KafkaEvent -> Bool

Show KafkaEvent

Defined in AWS.Lambda.Events.Kafka

Methods

showsPrec :: Int -> KafkaEvent -> ShowS

show :: KafkaEvent -> String

showList :: [KafkaEvent] -> ShowS

Generic KafkaEvent

Defined in AWS.Lambda.Events.Kafka

Associated Types

type Rep KafkaEvent :: Type -> Type

Methods

from :: KafkaEvent -> Rep KafkaEvent x

to :: Rep KafkaEvent x -> KafkaEvent

ToJSON KafkaEvent

Defined in AWS.Lambda.Events.Kafka

Methods

toJSON :: KafkaEvent -> Value

toEncoding :: KafkaEvent -> Encoding

toJSONList :: [KafkaEvent] -> Value

toEncodingList :: [KafkaEvent] -> Encoding

FromJSON KafkaEvent

Defined in AWS.Lambda.Events.Kafka

Methods

parseJSON :: Value -> Parser KafkaEvent

parseJSONList :: Value -> Parser [KafkaEvent]

type Rep KafkaEvent

Defined in AWS.Lambda.Events.Kafka

type Rep KafkaEvent = D1 ('MetaData "KafkaEvent" "AWS.Lambda.Events.Kafka" "hal-0.4.10-LWrkeryoXTOAPQgDhWLet9" 'False) (C1 ('MetaCons "KafkaEvent" 'PrefixI 'True) ((S1 ('MetaSel ('Just "eventSource") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 EventSource) :*: S1 ('MetaSel ('Just "eventSourceArn") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 (Maybe Text))) :*: (S1 ('MetaSel ('Just "bootstrapServers") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 (NonEmpty Text)) :*: S1 ('MetaSel ('Just "records") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 (Map Text [Record])))))

data EventSource

Constructors

AwsKafka

"aws:kafka"

SelfManagedKafka

SelfManagedKafka

Instances

Bounded EventSource

Defined in AWS.Lambda.Events.Kafka

Methods

minBound :: EventSource

maxBound :: EventSource

Enum EventSource

Defined in AWS.Lambda.Events.Kafka

Methods

succ :: EventSource -> EventSource

pred :: EventSource -> EventSource

toEnum :: Int -> EventSource

fromEnum :: EventSource -> Int

enumFrom :: EventSource -> [EventSource]

enumFromThen :: EventSource -> EventSource -> [EventSource]

enumFromTo :: EventSource -> EventSource -> [EventSource]

enumFromThenTo :: EventSource -> EventSource -> EventSource -> [EventSource]

Eq EventSource

Defined in AWS.Lambda.Events.Kafka

Methods

(==) :: EventSource -> EventSource -> Bool

(/=) :: EventSource -> EventSource -> Bool

Ord EventSource

Defined in AWS.Lambda.Events.Kafka

Methods

compare :: EventSource -> EventSource -> Ordering

(<) :: EventSource -> EventSource -> Bool

(<=) :: EventSource -> EventSource -> Bool

(>) :: EventSource -> EventSource -> Bool

(>=) :: EventSource -> EventSource -> Bool

max :: EventSource -> EventSource -> EventSource

min :: EventSource -> EventSource -> EventSource

Show EventSource

Defined in AWS.Lambda.Events.Kafka

Methods

showsPrec :: Int -> EventSource -> ShowS

show :: EventSource -> String

showList :: [EventSource] -> ShowS

Generic EventSource

Defined in AWS.Lambda.Events.Kafka

Associated Types

type Rep EventSource :: Type -> Type

Methods

from :: EventSource -> Rep EventSource x

to :: Rep EventSource x -> EventSource

ToJSON EventSource

Defined in AWS.Lambda.Events.Kafka

Methods

toJSON :: EventSource -> Value

toEncoding :: EventSource -> Encoding

toJSONList :: [EventSource] -> Value

toEncodingList :: [EventSource] -> Encoding

FromJSON EventSource

Defined in AWS.Lambda.Events.Kafka

Methods

parseJSON :: Value -> Parser EventSource

parseJSONList :: Value -> Parser [EventSource]

type Rep EventSource

Defined in AWS.Lambda.Events.Kafka

type Rep EventSource = D1 ('MetaData "EventSource" "AWS.Lambda.Events.Kafka" "hal-0.4.10-LWrkeryoXTOAPQgDhWLet9" 'False) (C1 ('MetaCons "AwsKafka" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "SelfManagedKafka" 'PrefixI 'False) (U1 :: Type -> Type))

type Record = Record' ByteString

Convenience alias: most of the time you will parse the records straight into some app-specific structure.

data Record' a

Records from a Kafka event. This is Traversable, which means you can do things like parse a JSON-encoded payload:

traverse decodeStrict :: FromJSON a => Record -> Maybe (Record' a)

Constructors

Record

Fields

• topic :: !Text
• partition :: !Int32
• offset :: !Int64
• timestamp :: !Timestamp
• headers :: [Header]

  NOTE: there can be multiple headers for a given key.

• key :: !(Maybe ByteString)
• value :: !(Maybe a)

Instances

Functor Record'

Defined in AWS.Lambda.Events.Kafka

Methods

fmap :: (a -> b) -> Record' a -> Record' b

(<$) :: a -> Record' b -> Record' a

Foldable Record'

Defined in AWS.Lambda.Events.Kafka

Methods

fold :: Monoid m => Record' m -> m

foldMap :: Monoid m => (a -> m) -> Record' a -> m

foldMap' :: Monoid m => (a -> m) -> Record' a -> m

foldr :: (a -> b -> b) -> b -> Record' a -> b

foldr' :: (a -> b -> b) -> b -> Record' a -> b

foldl :: (b -> a -> b) -> b -> Record' a -> b

foldl' :: (b -> a -> b) -> b -> Record' a -> b

foldr1 :: (a -> a -> a) -> Record' a -> a

foldl1 :: (a -> a -> a) -> Record' a -> a

toList :: Record' a -> [a]

null :: Record' a -> Bool

length :: Record' a -> Int

elem :: Eq a => a -> Record' a -> Bool

maximum :: Ord a => Record' a -> a

minimum :: Ord a => Record' a -> a

sum :: Num a => Record' a -> a

product :: Num a => Record' a -> a

Traversable Record'

Defined in AWS.Lambda.Events.Kafka

Methods

traverse :: Applicative f => (a -> f b) -> Record' a -> f (Record' b)

sequenceA :: Applicative f => Record' (f a) -> f (Record' a)

mapM :: Monad m => (a -> m b) -> Record' a -> m (Record' b)

sequence :: Monad m => Record' (m a) -> m (Record' a)

Eq a => Eq (Record' a)

Defined in AWS.Lambda.Events.Kafka

Methods

(==) :: Record' a -> Record' a -> Bool

(/=) :: Record' a -> Record' a -> Bool

Show a => Show (Record' a)

Defined in AWS.Lambda.Events.Kafka

Methods

showsPrec :: Int -> Record' a -> ShowS

show :: Record' a -> String

showList :: [Record' a] -> ShowS

Generic (Record' a)

Defined in AWS.Lambda.Events.Kafka

Associated Types

type Rep (Record' a) :: Type -> Type

Methods

from :: Record' a -> Rep (Record' a) x

to :: Rep (Record' a) x -> Record' a

ToJSON (Record' ByteString)

Encodes keys and values into base64.

Defined in AWS.Lambda.Events.Kafka

Methods

toJSON :: Record' ByteString -> Value

toEncoding :: Record' ByteString -> Encoding

toJSONList :: [Record' ByteString] -> Value

toEncodingList :: [Record' ByteString] -> Encoding

FromJSON (Record' ByteString)

Decodes base64-encoded keys and values, where present.

Defined in AWS.Lambda.Events.Kafka

Methods

parseJSON :: Value -> Parser (Record' ByteString)

parseJSONList :: Value -> Parser [Record' ByteString]

type Rep (Record' a)

Defined in AWS.Lambda.Events.Kafka

type Rep (Record' a) = D1 ('MetaData "Record'" "AWS.Lambda.Events.Kafka" "hal-0.4.10-LWrkeryoXTOAPQgDhWLet9" 'False) (C1 ('MetaCons "Record" 'PrefixI 'True) ((S1 ('MetaSel ('Just "topic") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Text) :*: (S1 ('MetaSel ('Just "partition") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Int32) :*: S1 ('MetaSel ('Just "offset") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Int64))) :*: ((S1 ('MetaSel ('Just "timestamp") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Timestamp) :*: S1 ('MetaSel ('Just "headers") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 [Header])) :*: (S1 ('MetaSel ('Just "key") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 (Maybe ByteString)) :*: S1 ('MetaSel ('Just "value") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 (Maybe a))))))

data Header

AWS serialises record headers to JSON as an array of objects. From their docs:

"headers":[{"headerKey":[104, 101, 97, 100, 101, 114, 86, 97, 108, 117, 101]}]

Note:

>>> map chr [104, 101, 97, 100, 101, 114, 86, 97, 108, 117, 101]
"headerValue"

Constructors

Header !Text !ByteString

Instances

Eq Header

Defined in AWS.Lambda.Events.Kafka

Methods

(==) :: Header -> Header -> Bool

(/=) :: Header -> Header -> Bool

Show Header

Defined in AWS.Lambda.Events.Kafka

Methods

showsPrec :: Int -> Header -> ShowS

show :: Header -> String

showList :: [Header] -> ShowS

Generic Header

Defined in AWS.Lambda.Events.Kafka

Associated Types

type Rep Header :: Type -> Type

Methods

from :: Header -> Rep Header x

to :: Rep Header x -> Header

ToJSON Header

Defined in AWS.Lambda.Events.Kafka

Methods

toJSON :: Header -> Value

toEncoding :: Header -> Encoding

toJSONList :: [Header] -> Value

toEncodingList :: [Header] -> Encoding

FromJSON Header

Defined in AWS.Lambda.Events.Kafka

Methods

parseJSON :: Value -> Parser Header

parseJSONList :: Value -> Parser [Header]

type Rep Header

Defined in AWS.Lambda.Events.Kafka

type Rep Header = D1 ('MetaData "Header" "AWS.Lambda.Events.Kafka" "hal-0.4.10-LWrkeryoXTOAPQgDhWLet9" 'False) (C1 ('MetaCons "Header" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Text) :*: S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 ByteString)))

data Timestamp

Kafka timestamp types, derived from the Java client's enum at: https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/record/TimestampType.java

Constructors

NoTimestampType

CreateTime !UTCTime

LogAppendTime !UTCTime

Instances

Eq Timestamp

Defined in AWS.Lambda.Events.Kafka

Methods

(==) :: Timestamp -> Timestamp -> Bool

(/=) :: Timestamp -> Timestamp -> Bool

Show Timestamp

Defined in AWS.Lambda.Events.Kafka

Methods

showsPrec :: Int -> Timestamp -> ShowS

show :: Timestamp -> String

showList :: [Timestamp] -> ShowS

Generic Timestamp

Defined in AWS.Lambda.Events.Kafka

Associated Types

type Rep Timestamp :: Type -> Type

Methods

from :: Timestamp -> Rep Timestamp x

to :: Rep Timestamp x -> Timestamp

type Rep Timestamp

Defined in AWS.Lambda.Events.Kafka

type Rep Timestamp = D1 ('MetaData "Timestamp" "AWS.Lambda.Events.Kafka" "hal-0.4.10-LWrkeryoXTOAPQgDhWLet9" 'False) (C1 ('MetaCons "NoTimestampType" 'PrefixI 'False) (U1 :: Type -> Type) :+: (C1 ('MetaCons "CreateTime" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 UTCTime)) :+: C1 ('MetaCons "LogAppendTime" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 UTCTime))))

Internal

parseTimestamp :: Object -> Parser Timestamp

unparseTimestamp :: KeyValue kv => Timestamp -> [kv]

int64ToUTCTime :: Int64 -> UTCTime

utcTimeToInt64 :: UTCTime -> Int64