TsgcWSPClient_Kafka talks to Apache Kafka over the broker's own binary protocol on port 9092. There is no REST proxy in front of it and no librdkafka underneath, so a Kafka producer or consumer is a single self-contained executable you can debug end to end.
Kafka is a plain-TCP protocol, so the carrier is a TsgcTCPClient. Everything else is configured through KafkaOptions.
TsgcWSPClient_Kafka
Apache Kafka wire protocol, v2 record batches
TCP (9092), TLS optional
Delphi, C++ Builder
Transport: plain TCP and TLS. sgcMQ connects over plain TCP and TLS, which is all the Kafka wire protocol asks for. The other sgcMQ protocols do define WebSocket transports, for example MQTT over WebSocket or AMQP over WebSocket, and those require a sgcWebSockets package, which provides the WebSocket client.
Set a GroupId and the component joins a consumer group, receives a partition assignment and commits offsets for you. Leave it empty to read partitions directly with no group at all.
uses
sgcTCP_Client_WS, sgcWebSocket_Protocols, sgcKafka_Classes;
var
TCPClient: TsgcTCPClient;
Kafka: TsgcWSPClient_Kafka;
begin
Kafka := TsgcWSPClient_Kafka.Create(nil);
Kafka.KafkaOptions.ClientId := 'my-delphi-app';
Kafka.KafkaOptions.Producer.Acks := kafkaAcksLeader;
Kafka.KafkaOptions.Producer.Compression := kafkaCompressionGzip;
Kafka.KafkaOptions.Consumer.GroupId := 'my-group';
Kafka.KafkaOptions.Consumer.OffsetReset := kafkaOffsetEarliest;
Kafka.OnKafkaMessage := KafkaMessage;
Kafka.OnKafkaProduce := KafkaProduce;
TCPClient := TsgcTCPClient.Create(nil);
Kafka.Client := TCPClient;
TCPClient.Host := '127.0.0.1';
TCPClient.Port := 9092;
TCPClient.Active := True;
// produce a record to a topic
Kafka.Produce('my-topic', 'Hello Kafka', 'key-1');
// consume: subscribe once, then Poll repeatedly (e.g. from a timer)
Kafka.Subscribe(['my-topic']);
end;
procedure TForm1.KafkaMessage(Sender: TObject;
const Message: TsgcKafkaMessage);
begin
Memo1.Lines.Add(Message.GetKeyString + ' = ' + Message.GetValueString);
end;
// fetch records, commit when a batch is processed
var
Messages: TsgcKafkaMessages;
begin
Messages := Kafka.Poll(1000);
try
if Messages.Count > 0 then
Kafka.CommitSync;
finally
Messages.Free;
end;
end;// include: sgcTCP_Client_WS.hpp, sgcWebSocket_Protocols.hpp,
// sgcKafka_Classes.hpp
TsgcWSPClient_Kafka *Kafka = new TsgcWSPClient_Kafka(this);
Kafka->KafkaOptions->ClientId = "my-cbuilder-app";
Kafka->KafkaOptions->Producer->Acks = kafkaAcksLeader;
Kafka->KafkaOptions->Consumer->GroupId = "my-group";
Kafka->KafkaOptions->Consumer->OffsetReset = kafkaOffsetEarliest;
Kafka->OnKafkaMessage = KafkaMessage;
TsgcTCPClient *TCPClient = new TsgcTCPClient(this);
Kafka->Client = TCPClient;
TCPClient->Host = "127.0.0.1";
TCPClient->Port = 9092;
TCPClient->Active = true;
Kafka->Produce("my-topic", "Hello Kafka", "key-1");
Kafka->Subscribe(ARRAYOFCONST(("my-topic")));
TsgcKafkaMessages *Messages = Kafka->Poll(1000);
try {
if (Messages->Count > 0)
Kafka->CommitSync();
}
__finally {
delete Messages;
}The members you reach for most often.
Produce(topic, value, key, partition) writes one record, ProduceBytes takes TBytes for key and value, and ProduceMessages sends a whole batch to one partition and returns the broker response.
KafkaOptions.Producer.Acks chooses kafkaAcksNone, kafkaAcksLeader or kafkaAcksAll. Compression switches on gzip, and TimeoutMs bounds the broker's ack wait.
Subscribe([topics]) registers interest and Poll(timeoutMs) fetches. Poll returns a TsgcKafkaMessages list that you own and must free, and also raises OnKafkaMessage per record.
Setting Consumer.GroupId turns on coordinator discovery, join and sync, partition assignment and background heartbeats. OnKafkaRebalance reports each assignment change.
CommitSync and CommitOffset commit explicitly, or set Consumer.AutoCommit with an interval. GetEarliestOffset, GetLatestOffset and GetCommittedOffset read positions back.
FetchMessages(topic, partition, offset, maxBytes) reads a specific partition from a specific offset, bypassing the consumer group entirely.
Consumer.MinBytes, MaxBytes, MaxPartitionBytes and MaxWaitMs trade latency against batch size, and SessionTimeoutMs and RebalanceTimeoutMs govern group membership.
CreateTopic takes a partition count and replication factor, DeleteTopic removes one, and GetMetadata, ListGroups and DescribeGroups inspect the cluster.
GetApiVersions asks the broker which protocol API versions it supports, which is the reliable way to branch on server version rather than guessing.