kafka

Dec 21, 2024

Changes in this version

type ConsumeConfig

+ ExpectTimeout bool

Apr 10, 2024

Changes in this version

type ConsumeConfig

+ NanoPrecision bool

Sep 14, 2023

Changes in this version

+ type Duration struct

+ func (d *Duration) UnmarshalJSON(b []byte) error

+ func (d Duration) MarshalJSON() ([]byte, error)

May 17, 2023

Changes in this version

+ var ErrNoJKSConfig = NewXk6KafkaError(failedConfigureJKS, "Failed to configure JKS", nil)

+ type JKS struct

+ ClientCertsPem []string

+ ClientKeyPem string

+ ServerCaPem string

+ type JKSConfig struct

+ ClientCertAlias string

+ ClientKeyAlias string

+ ClientKeyPassword string

+ Password string

+ Path string

+ ServerCaAlias string

Jan 23, 2023

Changes in this version

+ var ErrPartitionAndGroupID = NewXk6KafkaError(partitionAndGroupID, "Partition and groupID cannot be set at the same time", ...)

+ var ErrTopicAndGroupID = NewXk6KafkaError(topicAndGroupID, ...)

+ var HeartbeatInterval = time.Second * 3

+ var JoinGroupBackoff = time.Second * 5

+ var PartitionWatchInterval = time.Second * 5

+ var RetentionTime = time.Hour * 24

+ var SessionTimeout = time.Second * 30

+ var StartOffsets map[string]int64

type ReaderConfig

+ OffsetOutOfRangeError bool

+ ReadBatchTimeout time.Duration

Aug 12, 2022

Changes in this version

+ var ErrFailedTypeCast = NewXk6KafkaError(failedTypeCast, "Failed to cast type", nil)

+ var ErrInvalidSchema = NewXk6KafkaError(failedUnmarshalSchema, "Failed to unmarshal schema", nil)

+ var ErrNoSchemaRegistryClient = NewXk6KafkaError(failedConfigureSchemaRegistryClient, ...)

+ var ErrUnknownSerdesType = NewXk6KafkaError(invalidSerdeType, "Unknown serdes type", nil)

+ var ErrUnsupportedOperation = NewXk6KafkaError(unsupportedOperation, "Operation not supported", nil)

+ var TypesRegistry map[string]Serdes = map[string]Serdes

+ func GetSerdes(schemaType string) (Serdes, *Xk6KafkaError)

+ type AvroSerde struct

+ func (*AvroSerde) Deserialize(data []byte, schema *Schema) (interface{}, *Xk6KafkaError)

+ func (*AvroSerde) Serialize(data interface{}, schema *Schema) ([]byte, *Xk6KafkaError)

+ type ByteArraySerde struct

+ func (*ByteArraySerde) Deserialize(data []byte, schema *Schema) (interface{}, *Xk6KafkaError)

+ func (*ByteArraySerde) Serialize(data interface{}, schema *Schema) ([]byte, *Xk6KafkaError)

+ type Container struct

+ Data interface{}

+ Schema *Schema

+ SchemaType string

+ type JSONSerde struct

+ func (*JSONSerde) Deserialize(data []byte, schema *Schema) (interface{}, *Xk6KafkaError)

+ func (*JSONSerde) Serialize(data interface{}, schema *Schema) ([]byte, *Xk6KafkaError)

+ type Schema struct

+ EnableCaching bool

+ ID int

+ References []srclient.Reference

+ Schema string

+ SchemaType *srclient.SchemaType

+ Subject string

+ Version int

+ func (s *Schema) Codec() *goavro.Codec

+ func (s *Schema) JsonSchema() *jsonschema.Schema

+ type SchemaRegistryConfig struct

+ BasicAuth BasicAuth

+ EnableCaching bool

+ TLS TLSConfig

+ URL string

+ type Serdes interface

+ Deserialize func(data []byte, schema *Schema) (interface{}, *Xk6KafkaError)

+ Serialize func(data interface{}, schema *Schema) ([]byte, *Xk6KafkaError)

+ type StringSerde struct

+ func (*StringSerde) Deserialize(data []byte, schema *Schema) (interface{}, *Xk6KafkaError)

+ func (*StringSerde) Serialize(data interface{}, schema *Schema) ([]byte, *Xk6KafkaError)

+ type SubjectNameConfig struct

+ Element Element

+ Schema string

+ SubjectNameStrategy string

+ Topic string

+ type WireFormat struct

+ Data []byte

+ SchemaID int

Jul 13, 2022

Changes in this version

+ const ConcurrentRequests

+ const JSONSchemaDeserializer

+ const JSONSchemaSerializer

+ const MagicPrefixSize

+ const Timeout

+ var ErrForbiddenInInitContext = NewXk6KafkaError(kafkaForbiddenInInitContext, ...)

+ var ErrInvalidDataType = NewXk6KafkaError(invalidDataType, "Invalid data type provided for serializer/deserializer", nil)

+ var ErrInvalidPEMData = errors.New("tls: failed to find any PEM data in certificate input")

+ var ErrNotEnoughArguments = errors.New("not enough arguments")

+ var MaxWait = time.Millisecond * 200

+ var RebalanceTimeout = time.Second * 5

+ type Module struct

+ func (m *Module) Exports() modules.Exports

type SchemaRegistryConfiguration

+ URL string

type Xk6KafkaError

+ func DeserializeJSON(configuration Configuration, topic string, data []byte, element Element, ...) (interface{}, *Xk6KafkaError)

+ func SerializeJSON(configuration Configuration, topic string, data interface{}, element Element, ...) ([]byte, *Xk6KafkaError)

Jul 2, 2022

Changes in this version

+ const NONE

+ const SASL_PLAIN

+ var BALANCER_CRC32 = "balancer_crc32"

+ var BALANCER_HASH = "balancer_hash"

+ var BALANCER_LEAST_BYTES = "balancer_leastbytes"

+ var BALANCER_MURMUR2 = "balancer_murmur2"

+ var BALANCER_ROUND_ROBIN = "balancer_roundrobin"

+ var Balancers map[string]kafkago.Balancer

+ var CODEC_GZIP = "gzip"

+ var CODEC_LZ4 = "lz4"

+ var CODEC_SNAPPY = "snappy"

+ var CODEC_ZSTD = "zstd"

+ var GROUP_BALANCER_RACK_AFFINITY = "group_balancer_rack_affinity"

+ var GROUP_BALANCER_RANGE = "group_balancer_range"

+ var GROUP_BALANCER_ROUND_ROBIN = "group_balancer_round_robin"

+ var GroupBalancers map[string]kafkago.GroupBalancer

+ var ISOLATION_LEVEL_READ_COMMITTED = "isolation_level_read_committed"

+ var ISOLATION_LEVEL_READ_UNCOMMITTED = "isolation_level_read_uncommitted"

+ var IsolationLevels map[string]kafkago.IsolationLevel

+ var TLSVersions map[string]uint16

+ type ConnectionConfig struct

+ Address string

+ SASL SASLConfig

+ TLS TLSConfig

+ type ConsumeConfig struct

+ Config Configuration

+ KeySchema string

+ Limit int64

+ ValueSchema string

type Kafka

+ func (k *Kafka) XConnection(call goja.ConstructorCall) *goja.Object

+ func (k *Kafka) XReader(call goja.ConstructorCall) *goja.Object

+ func (k *Kafka) XWriter(call goja.ConstructorCall) *goja.Object

+ type Message struct

+ Headers map[string]interface{}

+ HighWaterMark int64

+ Key string

+ Offset int64

+ Partition int

+ Time time.Time

+ Topic string

+ Value string

+ type ProduceConfig struct

+ Config Configuration

+ KeySchema string

+ Messages []Message

+ ValueSchema string

+ type ReaderConfig struct

+ Brokers []string

+ CommitInterval time.Duration

+ ConnectLogger bool

+ GroupBalancers []string

+ GroupID string

+ GroupTopics []string

+ HeartbeatInterval time.Duration

+ IsolationLevel string

+ JoinGroupBackoff time.Duration

+ MaxAttempts int

+ MaxBytes int

+ MaxWait time.Duration

+ MinBytes int

+ Offset int64

+ Partition int

+ PartitionWatchInterval time.Duration

+ QueueCapacity int

+ ReadBackoffMax time.Duration

+ ReadBackoffMin time.Duration

+ ReadLagInterval time.Duration

+ RebalanceTimeout time.Duration

+ RetentionTime time.Duration

+ SASL SASLConfig

+ SessionTimeout time.Duration

+ StartOffset int64

+ TLS TLSConfig

+ Topic string

+ WatchPartitionChanges bool

type SchemaRegistryConfiguration

+ TLS TLSConfig

+ type WriterConfig struct

+ AutoCreateTopic bool

+ Balancer string

+ BatchBytes int

+ BatchSize int

+ BatchTimeout time.Duration

+ Brokers []string

+ Compression string

+ ConnectLogger bool

+ MaxAttempts int

+ ReadTimeout time.Duration

+ RequiredAcks int

+ SASL SASLConfig

+ TLS TLSConfig

+ Topic string

+ WriteTimeout time.Duration

Jun 20, 2022

Changes in this version

+ const RecordNameStrategy

+ const SASL_Plain

+ const SASL_SCRAM_SHA256

+ const SASL_SCRAM_SHA512

+ const SASL_SSL

+ const TLSv10

+ const TLSv11

+ const TLSv12

+ const TLSv13

+ const TopicNameStrategy

+ const TopicRecordNameStrategy

type ConsumerConfiguration

+ SubjectNameStrategy string

+ UseMagicPrefix bool

type Kafka

+ func (k *Kafka) GetKafkaControllerConnection(address string, saslConfig SASLConfig, tlsConfig TLSConfig) (*kafkago.Conn, *Xk6KafkaError)

type ProducerConfiguration

+ SubjectNameStrategy string

+ type SASLConfig struct

+ Algorithm string

+ Password string

+ Username string

type TLSConfig

+ EnableTLS bool

+ InsecureSkipTLSVerify bool

+ MinVersion string

type Xk6KafkaError

+ func GetDialer(saslConfig SASLConfig, tlsConfig TLSConfig) (*kafkago.Dialer, *Xk6KafkaError)

+ func GetSASLMechanism(saslConfig SASLConfig) (sasl.Mechanism, *Xk6KafkaError)

+ func GetSubjectName(schema string, topic string, element Element, subjectNameStrategy string) (string, *Xk6KafkaError)

May 31, 2022

Changes in this version

+ const AvroDeserializer

+ const AvroSerializer

+ const ByteArray

+ const ByteArrayDeserializer

+ const ByteArraySerializer

+ const JsonSchemaDeserializer

+ const JsonSchemaSerializer

+ const ProtobufDeserializer

+ const ProtobufSerializer

+ const String

+ const StringDeserializer

+ const StringSerializer

+ var DefaultDeserializer = StringDeserializer

+ var DefaultSerializer = StringSerializer

+ var ErrorForbiddenInInitContext = NewXk6KafkaError(kafkaForbiddenInInitContext, ...)

+ func EncodeWireFormat(data []byte, schemaID int) []byte

+ func FileExists(filename string) bool

+ func GivenCredentials(configuration Configuration) bool

+ func SchemaRegistryClientWithConfiguration(configuration SchemaRegistryConfiguration) *srclient.SchemaRegistryClient

+ func UnmarshalConfiguration(jsonConfiguration string) (Configuration, *Xk6KafkaError)

+ func UnmarshalCredentials(auth string) (*Credentials, *Xk6KafkaError)

type BasicAuth

+ Password string

+ Username string

type Credentials

+ TLSConfig *TLSConfig

+ type Element string

+ const Key

+ const Value

type Kafka

+ func (k *Kafka) GetDeserializer(schema string) Deserializer

+ func (k *Kafka) GetKafkaConnection(address, auth string) (*kafkago.Conn, *Xk6KafkaError)

+ func (k *Kafka) GetSerializer(schema string) Serializer

type SchemaRegistryConfiguration

+ TLSConfig *TLSConfig

+ type Serde struct

+ Registry map[string]*SerdeType[T]

+ func NewDeserializersRegistry() *Serde[Deserializer]

+ func NewSerializersRegistry() *Serde[Serializer]

+ type SerdeType struct

+ Class string

+ Function T

+ SchemaType srclient.SchemaType

+ WireFormatted bool

+ func NewSerdes(function T, class string, schemaType srclient.SchemaType, wireFormatted bool) *SerdeType[T]

+ func (s *SerdeType[Deserializer]) GetDeserializer() Deserializer

+ func (s *SerdeType[Serializer]) GetSerializer() Serializer

+ func (s *SerdeType[T]) GetSchemaType() srclient.SchemaType

+ func (s *SerdeType[T]) IsWireFormatted() bool

+ type TLSConfig struct

+ ClientCertPem string

+ ClientKeyPem string

+ ServerCaPem string

+ type Xk6KafkaError struct

+ Code errCode

+ Message string

+ OriginalError error

+ func CreateSchema(client *srclient.SchemaRegistryClient, subject string, schema string, ...) (*srclient.Schema, *Xk6KafkaError)

+ func DecodeWireFormat(message []byte) ([]byte, *Xk6KafkaError)

+ func DeserializeJson(configuration Configuration, topic string, data []byte, element Element, ...) (interface{}, *Xk6KafkaError)

+ func GetDialerFromAuth(auth string) (*kafkago.Dialer, *Xk6KafkaError)

+ func GetDialerFromCreds(creds *Credentials) (*kafkago.Dialer, *Xk6KafkaError)

+ func GetSchema(client *srclient.SchemaRegistryClient, subject string, schema string, ...) (*srclient.Schema, *Xk6KafkaError)

+ func GetTLSConfig(tlsConfig *TLSConfig) (*tls.Config, *Xk6KafkaError)

+ func NewXk6KafkaError(code errCode, msg string, originalErr error) *Xk6KafkaError

+ func SerializeJson(configuration Configuration, topic string, data interface{}, element Element, ...) ([]byte, *Xk6KafkaError)

+ func ValidateConfiguration(configuration Configuration) *Xk6KafkaError

+ func (e Xk6KafkaError) Error() string

+ func (e Xk6KafkaError) Unwrap() error

May 5, 2022

Changes in this version

+ const None

type Credentials

+ ClientCertPem string

+ ClientKeyPem string

+ ServerCaPem string

type Kafka

+ func (k *Kafka) DeleteTopic(address, topic string, auth string) error

+ type KafkaModule struct

+ func (c *KafkaModule) Exports() modules.Exports

+ type RootModule struct

+ func (*RootModule) NewModuleInstance(vu modules.VU) modules.Instance

+ type SchemaInfo struct

+ Id int32

+ Version int32

type SchemaRegistryConfiguration

+ UseLatest bool

Apr 22, 2022

Changes in this version

+ func DeserializeAvro(configuration Configuration, data []byte, keyOrValue string, schema string) interface

+ func DeserializeByteArray(configuration Configuration, data []byte, keyOrValue string, schema string) interface

+ func DeserializeString(configuration Configuration, data []byte, keyOrValue string, schema string) interface

+ func SerializeAvro(configuration Configuration, topic string, data interface{}, keyOrValue string, ...) ([]byte, error)

+ func SerializeByteArray(configuration Configuration, topic string, data interface{}, keyOrValue string, ...) ([]byte, error)

+ func SerializeString(configuration Configuration, topic string, data interface{}, keyOrValue string, ...) ([]byte, error)

+ type Deserializer func(configuration Configuration, data []byte, keyOrValue string, schema string) interface

+ func GetDeserializer(deserializer string, schema string) Deserializer

+ type Serializer func(configuration Configuration, topic string, data interface{}, keyOrValue string, ...) ([]byte, error)

+ func GetSerializer(serializer string, schema string) Serializer

Sep 6, 2021

Changes in this version

+ var CompressionCodecs = map[string]compress.Codec

Jul 4, 2021

Changes in this version

type Kafka

+ func (*Kafka) CreateTopic(address, topic string, partitions, replicationFactor int) error

+ func (*Kafka) ListTopics(address string) ([]string, error)

Jun 29, 2021

Changes in this version

+ func ConsumeInternal(ctx context.Context, reader *kafkago.Reader, limit int64, ...) []map[string]interface

+ func ProduceInternal(ctx context.Context, writer *kafkago.Writer, messages []map[string]string, ...) error

+ type BasicAuth struct

+ CredentialsSource string

+ UserInfo string

+ type Configuration struct

+ Consumer ConsumerConfiguration

+ Producer ProducerConfiguration

+ SchemaRegistry SchemaRegistryConfiguration

+ type ConsumerConfiguration struct

+ KeyDeserializer string

+ ValueDeserializer string

type Kafka

+ func (*Kafka) ConsumeWithConfiguration(ctx context.Context, reader *kafkago.Reader, limit int64, ...) []map[string]interface{}

+ func (*Kafka) ProduceWithConfiguration(ctx context.Context, writer *kafkago.Writer, messages []map[string]string, ...) error

+ type ProducerConfiguration struct

+ KeySerializer string

+ ValueSerializer string

+ type SchemaRegistryConfiguration struct

+ BasicAuth BasicAuth

+ Url string

Jun 18, 2021

Changes in this version

+ const Plain

+ const SHA256

+ const SHA512

+ type Credentials struct

+ Algorithm string

+ Password string

+ Username string

Nov 25, 2020

Changes in this version

+ type Kafka struct

+ func (*Kafka) Consume(ctx context.Context, reader *kafkago.Reader, limit int64, keySchema string, ...) []map[string]interface{}

+ func (*Kafka) Produce(ctx context.Context, writer *kafkago.Writer, messages []map[string]string, ...) error

+ func (*Kafka) Reader(brokers []string, topic string, partition int, minBytes int, maxBytes int, ...) *kafkago.Reader

+ func (*Kafka) Writer(brokers []string, topic string) *kafkago.Writer

Jul 26, 2020

Changes in this version

+ var ReaderBytes = stats.New("kafka.reader.message.bytes", stats.Counter, stats.Data)

+ var ReaderDialTime = stats.New("kafka.reader.dial.seconds", stats.Trend, stats.Time)

+ var ReaderDials = stats.New("kafka.reader.dial.count", stats.Counter)

+ var ReaderErrors = stats.New("kafka.reader.error.count", stats.Counter)

+ var ReaderFetchBytes = stats.New("kafka.reader.fetch.bytes", stats.Counter, stats.Data)

+ var ReaderFetchSize = stats.New("kafka.reader.fetch.size", stats.Counter)

+ var ReaderFetches = stats.New("kafka.reader.fetches.count", stats.Counter)

+ var ReaderLag = stats.New("kafka.reader.lag", stats.Gauge)

+ var ReaderMaxBytes = stats.New("kafka.reader.fetch_bytes.max", stats.Gauge)

+ var ReaderMaxWait = stats.New("kafka.reader.fetch_wait.max", stats.Gauge, stats.Time)

+ var ReaderMessages = stats.New("kafka.reader.message.count", stats.Counter)

+ var ReaderMinBytes = stats.New("kafka.reader.fetch_bytes.min", stats.Gauge)

+ var ReaderOffset = stats.New("kafka.reader.offset", stats.Gauge)

+ var ReaderQueueCapacity = stats.New("kafka.reader.queue.capacity", stats.Gauge)

+ var ReaderQueueLength = stats.New("kafka.reader.queue.length", stats.Gauge)

+ var ReaderReadTime = stats.New("kafka.reader.read.seconds", stats.Trend, stats.Time)

+ var ReaderRebalances = stats.New("kafka.reader.rebalance.count", stats.Counter)

+ var ReaderTimeouts = stats.New("kafka.reader.timeouts.count", stats.Counter)

+ var ReaderWaitTime = stats.New("kafka.reader.wait.seconds", stats.Trend, stats.Time)

+ var WriterAsync = stats.New("kafka.writer.async", stats.Rate)

+ var WriterBatchBytes = stats.New("kafka.writer.batch.bytes", stats.Counter, stats.Data)

+ var WriterBatchSize = stats.New("kafka.writer.batch.size", stats.Counter)

+ var WriterBatchTimeout = stats.New("kafka.writer.batch.timeout", stats.Gauge, stats.Time)

+ var WriterBytes = stats.New("kafka.writer.message.bytes", stats.Counter, stats.Data)

+ var WriterDialTime = stats.New("kafka.writer.dial.seconds", stats.Trend, stats.Time)

+ var WriterDials = stats.New("kafka.writer.dial.count", stats.Counter)

+ var WriterErrors = stats.New("kafka.writer.error.count", stats.Counter)

+ var WriterMaxAttempts = stats.New("kafka.writer.attempts.max", stats.Gauge)

+ var WriterMaxBatchSize = stats.New("kafka.writer.batch.max", stats.Gauge)

+ var WriterMessages = stats.New("kafka.writer.message.count", stats.Counter)

+ var WriterQueueCapacity = stats.New("kafka.writer.queue.capacity", stats.Gauge)

+ var WriterQueueLength = stats.New("kafka.writer.queue.length", stats.Gauge)

+ var WriterReadTimeout = stats.New("kafka.writer.read.timeout", stats.Gauge, stats.Time)

+ var WriterRebalanceInterval = stats.New("kafka.writer.rebalance.interval", stats.Gauge, stats.Time)

+ var WriterRebalances = stats.New("kafka.writer.rebalance.count", stats.Counter)

+ var WriterRequiredAcks = stats.New("kafka.writer.acks.required", stats.Gauge)

+ var WriterRetries = stats.New("kafka.writer.retries.count", stats.Counter)

+ var WriterWaitTime = stats.New("kafka.writer.wait.seconds", stats.Trend, stats.Time)

+ var WriterWriteTime = stats.New("kafka.writer.write.seconds", stats.Trend, stats.Time)

+ var WriterWriteTimeout = stats.New("kafka.writer.write.timeout", stats.Gauge, stats.Time)

+ var WriterWrites = stats.New("kafka.writer.write.count", stats.Counter)

+ func FromAvro(message []byte, schema string) interface

+ func ReportError(err error, msg string)

+ func ReportReaderStats(ctx context.Context, currentStats kafkago.ReaderStats) error

+ func ReportWriterStats(ctx context.Context, currentStats kafkago.WriterStats) error

+ func ToAvro(value string, schema string) []byte

Jul 24, 2020

Changes in this version

+ var Async = stats.New("kafka.writer.async", stats.Rate)

+ var BatchBytes = stats.New("kafka.writer.batch.bytes", stats.Counter, stats.Data)

+ var BatchSize = stats.New("kafka.writer.batch.size", stats.Counter)

+ var BatchTimeout = stats.New("kafka.writer.batch.timeout", stats.Gauge, stats.Time)

+ var Bytes = stats.New("kafka.writer.message.bytes", stats.Counter, stats.Data)

+ var DialTime = stats.New("kafka.writer.dial.seconds", stats.Trend, stats.Time)

+ var Dials = stats.New("kafka.writer.dial.count", stats.Counter)

+ var Errors = stats.New("kafka.writer.error.count", stats.Counter)

+ var JavaScriptPlugin plugin

+ var MaxAttempts = stats.New("kafka.writer.attempts.max", stats.Gauge)

+ var MaxBatchSize = stats.New("kafka.writer.batch.max", stats.Gauge)

+ var Messages = stats.New("kafka.writer.message.count", stats.Counter)

+ var QueueCapacity = stats.New("kafka.writer.queue.capacity", stats.Gauge)

+ var QueueLength = stats.New("kafka.writer.queue.length", stats.Gauge)

+ var ReadTimeout = stats.New("kafka.writer.read.timeout", stats.Gauge, stats.Time)

+ var RebalanceInterval = stats.New("kafka.writer.rebalance.interval", stats.Gauge, stats.Time)

+ var Rebalances = stats.New("kafka.writer.rebalance.count", stats.Counter)

+ var RequiredAcks = stats.New("kafka.writer.acks.required", stats.Gauge)

+ var Retries = stats.New("kafka.writer.retries.count", stats.Counter)

+ var WaitTime = stats.New("kafka.writer.wait.seconds", stats.Trend, stats.Time)

+ var WriteTime = stats.New("kafka.writer.write.seconds", stats.Trend, stats.Time)

+ var WriteTimeout = stats.New("kafka.writer.write.timeout", stats.Gauge, stats.Time)

+ var Writes = stats.New("kafka.writer.write.count", stats.Counter)

+ func GetState(ctx interface{}) (*lib.State, error)

+ func New() *kafka