tamediadigital / kafka-d Goto Github PK

View Code? Open in Web Editor NEW

15.0 15.0 10.0 126 KB

kafka-d is a D Kafka client that depends on the vibe.d framework.

License: MIT License

D 100.00%

kafka-d's People

Contributors

Stargazers

Watchers

Forkers

dkdewitt rjmcguire martinnowak duselbaer burner aliak00 madkote singingbush

kafka-d's Issues

support newer message format with lz4 compression

I've been taking a look at the source in consumer.d within processBuffer. It seems kafka-d only handles the older message format (prior to Kafka 0.11) which are structured like:

MessageSet (Version: 0) => [offset message_size message]
offset => INT64
message_size => INT32
message => crc magic_byte attributes key value
    crc => INT32
    magic_byte => INT8
    attributes => INT8
        bit 0~2:
            0: no compression
            1: gzip
            2: snappy
        bit 3~7: unused
    key => BYTES
    value => BYTES

newer versions of Kafka use the following structure which includes a timestamp as well as also supporting lz4 compression.

MessageSet (Version: 1) => [offset message_size message]
offset => INT64
message_size => INT32
message => crc magic_byte attributes timestamp key value
    crc => INT32
    magic_byte => INT8
    attributes => INT8
        bit 0~2:
            0: no compression
            1: gzip
            2: snappy
            3: lz4
        bit 3: timestampType
            0: create time
            1: log append time
        bit 4~7: unused
    timestamp => INT64
    key => BYTES
    value => BYTES

doesn't compile with ldc2

source/kafkad/consumer/consumer.d(21): Error: cannot create a struct until its size is determined
source/kafkad/consumer/consumer.d(22): Error: cannot create a struct until its size is determined

support auto creating a topic if it doesn't exist

When integrating with Kafka using Spring Framework it's easy to get a producer/consumer running without needing to configure anything and when running a producer, if the topic doesn't already exist, Spring will just create it. It's also configurable via auto.create.topics.enable which defaults to true.

It would be great if kafka-d could do the same as it really simplifies deploying a new producer

concurrency changes in vibe.d

vibe-d/vibe.d#1345
makes kafka-d not work anymore

Lack for consumer group ID functionality?

Didn't find any possibility to specify consumer group id for connection to use Kafka consumer message share mode (see https://www.tutorialspoint.com/apache_kafka/apache_kafka_workflow.htm, Workflow of Queue Messaging / Consumer Group)

Hope I missed something?

NotLeaderForPartition handling not working

https://github.com/tamediadigital/kafka-d/blob/master/source/kafkad/connection.d#L259
probably because of the mutex

v2.071.0 warnings

fix all module/package visibility warnings

ldc2 0.16.0 issues

0.16.0 beta is out: http://forum.dlang.org/post/[email protected]

todo:

remove pragma(inline)
fix nogc exception

submodules/kafka-d/source/kafkad/exception.d(7): Error: @nogc function 'kafkad.exception.ConnectionException.ExceptionCtorMixin!().this' cannot call non-@nogc function 'object.Exception.this'
submodules/kafka-d/source/kafkad/exception.d(43): Error: function kafkad.exception.rethrow!(ConnectionException, NetworkAddress, string).rethrow no return exp; or assert(0); at end of function
submodules/kafka-d/source/kafkad/client.d(148): Error: template instance kafkad.exception.rethrow!(ConnectionException, NetworkAddress, string) error instantiating
submodules/kafka-d/source/kafkad/exception.d(7): Error: @nogc function 'kafkad.exception.ProtocolException.ExceptionCtorMixin!().this' cannot call non-@nogc function 'object.Exception.this'
submodules/kafka-d/source/kafkad/exception.d(7): Error: @nogc function 'kafkad.exception.CrcException.ExceptionCtorMixin!().this' cannot call non-@nogc function 'object.Exception.this'
submodules/kafka-d/source/kafkad/consumer.d(78): Error: function kafkad.consumer.Consumer.getMessage no return exp; or assert(0); at end of function
submodules/kafka-d/source/kafkad/exception.d(10): Error: @nogc function 'kafkad.exception.ConnectionException.ExceptionCtorMixin!().this' cannot call non-@nogc function 'object.Exception.this'
submodules/kafka-d/source/kafkad/exception.d(7): Error: @nogc function 'kafkad.exception.StreamException.ExceptionCtorMixin!().this' cannot call non-@nogc function 'object.Exception.this'
submodules/kafka-d/source/kafkad/exception.d(10): Error: @nogc function 'kafkad.exception.StreamException.ExceptionCtorMixin!().this' cannot call non-@nogc function 'object.Exception.this'
submodules/kafka-d/source/kafkad/exception.d(7): Error: @nogc function 'kafkad.exception.MetadataException.ExceptionCtorMixin!().this' cannot call non-@nogc function 'object.Exception.this'
submodules/kafka-d/source/kafkad/exception.d(10): Error: @nogc function 'kafkad.exception.MetadataException.ExceptionCtorMixin!().this' cannot call non-@nogc function 'object.Exception.this'
submodules/kafka-d/source/kafkad/exception.d(10): Error: @nogc function 'kafkad.exception.ProtocolException.ExceptionCtorMixin!().this' cannot call non-@nogc function 'object.Exception.this'
submodules/kafka-d/source/kafkad/exception.d(10): Error: @nogc function 'kafkad.exception.CrcException.ExceptionCtorMixin!().this' cannot call non-@nogc function 'object.Exception.this'
submodules/kafka-d/source/kafkad/protocol/deserializer.d(67): Error: unrecognized pragma(inline)
submodules/kafka-d/source/kafkad/protocol/serializer.d(35): Error: unrecognized pragma(inline)