This is version 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 . It is not the current version, and thus it cannot be edited.
[Back to current version]   [Restore this version]

Kafka task#

Apache Kafka is a high-performance, open-source distributed event streaming platform developed by the Apache Software Foundation. It is designed for building real-time data pipelines and stream-processing applications that handle high-throughput, fault-tolerant, and scalable message flows across systems. (More info : Apache Kafak Link)

---

⚠️ Download the required JAR files and copy them into the CrushFTP/plugins/lib/ directory:
Download kafka-clients.jar version 4.0.0 Link
Download kafka-streams.jar version 4.0.0 Link
After placing the files, restart the CrushFTP service to apply the changes.
More info:
• Kafka Broker Version 4.0.x
• Minimum Java Version: Java 11 (supports 11, 17, 23)

---

The Kafka Task allows you to send either file contents or custom messages to a specified Kafka topic.



To use custom SASL connection settings, enable the Load custom client config flag. This allows you to provide your own client configuration properties.



Custom SASL Config example:

#KafkaTask SASL properties file example

# Kafka server host and port.
bootstrap.servers=192.168.0.10:9092

# Client id
client.id=MyCrushFTP

# Serializer (Do not change!)
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer

#Max block and timeout
max.block.ms=10000
request.timeout.ms=20000 

#Authentication related settings
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
    username="kafka-username" \
    password="kafka-password";
#Optional
acks=all
retries=3
metadata.max.age.ms=3000

Documentation Links:
• Apache Kafka – Security (SASL) Link
• JAAS Configuration for Kafka Clients Link

KafkaTask SASL properties Explained:

# Kafka server host and port:
• ⚠️ Specifies the address of the Kafka broker to connect to.
• You can list multiple brokers separated by commas for failover: 192.168.0.10:9092,192.168.0.11:9092

# Client id:
• A unique identifier for the client. This is useful for logging and monitoring within Kafka.

# Serializer (Do not change!):
• These specify how the key and value of each Kafka message are serialized.
• Key serializer: Converts the message key to a string.
• Value serializer: Converts the message content (e.g. a file) into a byte array.
• ⚠️ Do not change!

#Max block and timeout:
• max.block.ms: Maximum time (in milliseconds) a send call will block if the buffer is full. Here, 10 seconds.
• request.timeout.ms: Time before a request is considered failed due to no response. Set to 20 seconds here.

#Authentication related settings:
• security.protocol: Sets how the client communicates with Kafka.
• SASL_PLAINTEXT means SASL authentication over a plaintext (non-encrypted) channel.
• Use SASL_SSL if you want to secure the connection.
• sasl.mechanism: The SASL mechanism used. PLAIN is username/password-based authentication.

sasl.jaas.config: • sasl.jaas.config: The JAAS (Java Authentication and Authorization Service) configuration line to authenticate with Kafka using SASL/PLAIN.
• username / password: ⚠️ Replace these with your actual Kafka credentials.

Optioanal:

acks=all: The producer will wait for all in-sync replicas (ISRs) to acknowledge the record before considering the write successful. Ensures that data isn’t lost if a broker crashes. Slightly increases latency compared to acks=1 or acks=0.
retries=3: The producer will retry sending a record up to 3 times if a transient failure occurs (e.g., timeout, disconnect).
metadata.max.age.ms=3000: The producer will refresh its cached metadata (e.g., topic partitions, broker info) every 3 seconds. Lower values increase responsiveness but may cause more overhead due to more frequent metadata fetches. Usually, it is 300,000 ms (5 minutes), so 3,000 ms is very aggressive and useful only in dynamic environments.