### Start PubSub+ Kafka Sink Connector in Standalone Mode

Source: https://github.com/solaceproducts/pubsubplus-connector-kafka-sink/blob/master/README.md

This command starts the PubSub+ Sink Connector in standalone mode, using the Kafka standalone properties file and the PubSub+ Sink Connector properties file for configuration. Ensure the paths to these files are correct.

```shell
bin/connect-standalone.sh \
config/connect-standalone.properties \
connectors/pubsubplus-connector-kafka-sink-<version>/etc/solace_sink.properties
```

--------------------------------

### Run Kafka Connect in Standalone Mode with Solace Sink Connector

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

This section shows how to start the Kafka Connect service in standalone mode, enabling the Solace Sink Connector. It includes commands for both foreground and daemon execution, along with instructions on how to verify the connection and test message production.

```bash
# Start Kafka Connect in standalone mode with the Solace Sink Connector
bin/connect-standalone.sh \
  config/connect-standalone.properties \
  connectors/pubsubplus-connector-kafka-sink-3.1.0/etc/solace_sink.properties

# Run in daemon mode (logs to file instead of console)
bin/connect-standalone.sh -daemon \
  config/connect-standalone.properties \
  connectors/pubsubplus-connector-kafka-sink-3.1.0/etc/solace_sink.properties

# Verify connection in logs - look for:
# ================ JCSMPSession Connected

# Test by producing a message to Kafka
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
> Hello from Kafka!
> {"orderId": "12345", "product": "widget", "quantity": 10}

# The messages will appear on PubSub+ topic 'sinktest' (as configured in sol.topics)
```

--------------------------------

### Deploy PubSub+ Sink Connector via REST API (Shell)

Source: https://github.com/solaceproducts/pubsubplus-connector-kafka-sink/blob/master/README.md

This command initiates the deployment of the PubSub+ Sink Connector in distributed mode by sending a POST request to the Kafka Connect REST API. It uses a local JSON file containing the connector's configuration. Ensure the IP address and port match your Kafka Connect worker setup.

```shell
curl -X POST -H "Content-Type: application/json" \
             -d @solace_sink_properties.json \
             http://18.218.82.209:8083/connectors
```

--------------------------------

### SolDynamicDestinationRecordProcessor: Content-Based Routing (Java)

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

Enables dynamic routing of Kafka records to different Solace topics based on message content. This processor supports content-based routing patterns by inspecting the message payload. The example demonstrates routing based on commands like 'stop' and 'start', and also shows how the dynamic destination is stored in user properties.

```properties
name=dynamicSinkConnector
connector.class=com.solace.connector.kafka.connect.sink.SolaceSinkConnector
tasks.max=1
topics=bus-commands
sol.host=tcp://iot-broker.example.com:55555
sol.username=iot-connector
sol.password=iot-pass
sol.vpn_name=iot-vpn
sol.record_processor_class=com.solace.connector.kafka.connect.sink.recordprocessor.SolDynamicDestinationRecordProcessor
sol.dynamic_destination=true
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
```

```text
Kafka message: "B001 stop engine"
Solace topic: ctrl/bus/B001/stop
Payload: "stop engine"

Kafka message: "B002 start route"
Solace topic: ctrl/bus/B002/start
Payload: "start route"

Kafka message: "B003 passenger count 45"
Solace topic: comms/bus/B003
Payload: "passenger count 45"
```

--------------------------------

### Configure Kerberos Authentication

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

Enables Kerberos (GSS-KRB) authentication for secure communication with the Solace broker in enterprise environments. Requires Kerberos configuration files and principal setup.

```properties
# Kerberos authentication configuration
name=kerberosSinkConnector
connector.class=com.solace.connector.kafka.connect.sink.SolaceSinkConnector
tasks.max=1
topics=enterprise-events

# Solace connection (use DNS hostname, not IP)
sol.host=tcp://solace-broker.corp.example.com:55555
sol.username=kafka-connector@CORP.EXAMPLE.COM
sol.vpn_name=enterprise-vpn
sol.topics=enterprise/events

# Enable Kerberos authentication scheme
sol.authentication_scheme=AUTHENTICATION_SCHEME_GSS_KRB

# Kerberos configuration files (must exist on all worker nodes)
sol.kerberos.krb5.conf=/opt/kerberos/krb5.conf
sol.kerberos.login.conf=/opt/kerberos/login.conf

# Optional: Specify Kerberos service name (default: solace)
sol.krb_service_name=solace

# Record processor
sol.record_processor_class=com.solace.connector.kafka.connect.sink.recordprocessor.SolSimpleRecordProcessor
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
```

--------------------------------

### SolSimpleKeyedRecordProcessor: Kafka Key to Solace Message Header Mapping (Java)

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

Processes Kafka records that include keys, mapping the Kafka key to Solace message headers like CorrelationId. It supports different key handling modes: NONE, DESTINATION, CORRELATION_ID, and CORRELATION_ID_AS_BYTES, allowing flexible integration with Solace messaging patterns. The example shows configuration for mapping the key to Solace CorrelationId as a string.

```properties
name=keyedSinkConnector
connector.class=com.solace.connector.kafka.connect.sink.SolaceSinkConnector
tasks.max=2
topics=customer-events
sol.host=tcp://broker.example.com:55555
sol.username=kafka-user
sol.password=kafka-pass
sol.vpn_name=production
sol.topics=events/customers
sol.record_processor_class=com.solace.connector.kafka.connect.sink.recordprocessor.SolSimpleKeyedRecordProcessor
sol.kafka_message_key=CORRELATION_ID
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
key.converter=org.apache.kafka.connect.storage.StringConverter
```

```bash
bin/kafka-console-producer.sh --broker-list localhost:9092 \
  --topic customer-events \
  --property "parse.key=true" \
  --property "key.separator=:"
> customer-123:{"event":"purchase","amount":99.99}
> customer-456:{"event":"login","ip":"192.168.1.1"}
```

--------------------------------

### Build Solace Kafka Sink Connector from Source with Gradle

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

Provides bash commands to clone the connector's Git repository, initialize submodules, build the connector distributable archives using Gradle, and run tests. It also shows how to add the connector as a dependency in Maven and Gradle projects.

```bash
# Clone the repository
git clone https://github.com/SolaceProducts/pubsubplus-connector-kafka-sink.git
cd pubsubplus-connector-kafka-sink

# Initialize test support submodule
git submodule update --init --recursive
cd solace-integration-test-support
./mvnw clean install -DskipTests -Dchangelist=
cd ..

# Build the connector (creates distributable archives)
./gradlew clean build

# Build artifacts are created in:
# build/distributions/pubsubplus-connector-kafka-sink-<version>.zip
# build/distributions/pubsubplus-connector-kafka-sink-<version>.tar

# Run unit tests only
./gradlew test

# Run full integration tests (requires Docker)
# Spins up PubSub+ broker, Zookeeper, Kafka, and Kafka Connect
./gradlew clean test integrationTest

# Add custom record processor to project dependencies (Maven)
# <dependency>
#   <groupId>com.solace.connector.kafka.connect</groupId>
#   <artifactId>pubsubplus-connector-kafka-sink</artifactId>
#   <version>3.1.0</version>
# </dependency>

# Add custom record processor to project dependencies (Gradle)
# compile "com.solace.connector.kafka.connect:pubsubplus-connector-kafka-sink:3.1.0"
```

--------------------------------

### Produce Kafka Message with Headers using kcat

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

Demonstrates how to produce a Kafka message with custom headers using the `kcat` (formerly `kafkacat`) command-line tool. The output shows the command and the expected Solace message user properties that will be generated from these headers.

```bash
# Produce message with headers using kafkacat/kcat
echo '{"event":"test"}' | kcat -b localhost:9092 -t events-with-headers \
  -H "trace-id=abc123" \
  -H "source-system=crm" \
  -H "priority=high"

# Resulting Solace message user properties will include:
# - k_topic: events-with-headers
# - k_partition: 0
# - k_offset: 42
# - trace-id: abc123
# - source-system: crm
# - priority: high
```

--------------------------------

### Check Available Kafka Connectors (Shell)

Source: https://github.com/solaceproducts/pubsubplus-connector-kafka-sink/blob/master/README.md

This command verifies if the PubSub+ Sink Connector is registered and available in a distributed Kafka Connect environment. It queries the Kafka Connect REST API for connector plugins. The output should list the connector's class, type, and version.

```shell
curl http://18.218.82.209:8083/connector-plugins | jq
```

--------------------------------

### Configure Solace Kafka Sink Connector (Properties File)

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

Defines Kafka Connect parameters and Solace-specific connection settings for the sink connector using a properties file. Supports basic authentication, TLS, and Kerberos. Includes settings for Kafka topics, Solace broker connection, destination topics/queues, record processors, and transaction settings.

```properties
# Kafka Connect basic parameters
name=solaceSinkConnector
connector.class=com.solace.connector.kafka.connect.sink.SolaceSinkConnector
tasks.max=1
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
key.converter=org.apache.kafka.connect.storage.StringConverter

# Consumer group for parallel task coordination
group.id=solSinkConnectorGroup

# Kafka source topics (comma-separated for multiple)
topics=orders,events

# PubSub+ broker connection settings
sol.host=tcp://broker.example.com:55555
sol.username=kafka-connector
sol.password=secret123
sol.vpn_name=production

# Destination PubSub+ topics (comma-separated)
sol.topics=kafka/orders,kafka/events

# Alternative: Send to a PubSub+ queue instead
#sol.queue=kafka-sink-queue

# Record processor selection
sol.record_processor_class=com.solace.connector.kafka.connect.sink.recordprocessor.SolSimpleRecordProcessor

# Transaction settings for guaranteed delivery
sol.use_transactions_for_queue=true
sol.use_transactions_for_topics=false
sol.autoflush.size=200

# Channel connection tuning
sol.channel_properties.connect_timout_in_millis=30000
sol.channel_properties.reconnect_retries=5
sol.channel_properties.reconnect_retry_wait_in_millis=3000
```

--------------------------------

### Kerberos krb5.conf Configuration

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

Defines the Kerberos realm, KDC, and domain-to-realm mappings for the client. This file is essential for Kerberos authentication.

```ini
[libdefaults]
    default_realm = CORP.EXAMPLE.COM
    dns_lookup_realm = false
    dns_lookup_kdc = false

[realms]
    CORP.EXAMPLE.COM = {
        kdc = kdc.corp.example.com
        admin_server = kdc.corp.example.com
    }

[domain_realm]
    .corp.example.com = CORP.EXAMPLE.COM
    corp.example.com = CORP.EXAMPLE.COM
```

--------------------------------

### Manage Solace Kafka Sink Connector with Kafka Connect REST API

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

Demonstrates using `curl` commands to interact with the Kafka Connect REST API for managing the Solace PubSub+ Sink Connector. Includes checking plugin availability, deploying, checking status, pausing, resuming, and deleting the connector.

```bash
# Check if PubSub+ Sink Connector plugin is available
curl -s http://localhost:8083/connector-plugins | jq '.[] | select(.class | contains("SolaceSinkConnector"))'
# Expected output:
# {
#   "class": "com.solace.connector.kafka.connect.sink.SolaceSinkConnector",
#   "type": "sink",
#   "version": "3.1.0"
# }

# Deploy the connector
curl -X POST -H "Content-Type: application/json" \
  -d @solace_sink_properties.json \
  http://localhost:8083/connectors

# Check connector status
curl -s http://localhost:8083/connectors/solaceSinkConnector/status | jq
# Expected output:
# {
#   "name": "solaceSinkConnector",
#   "connector": {
#     "state": "RUNNING",
#     "worker_id": "worker1:8083"
#   },
#   "tasks": [
#     {
#       "id": 0,
#       "state": "RUNNING",
#       "worker_id": "worker1:8083"
#     }
#   ],
#   "type": "sink"
# }

# Pause the connector
curl -X PUT http://localhost:8083/connectors/solaceSinkConnector/pause

# Resume the connector
curl -X PUT http://localhost:8083/connectors/solaceSinkConnector/resume

# Delete the connector
curl -X DELETE http://localhost:8083/connectors/solaceSinkConnector
```

--------------------------------

### Forking a Git Repository

Source: https://github.com/solaceproducts/pubsubplus-connector-kafka-sink/blob/master/CONTRIBUTING.md

This command clones a forked version of a Git repository to your local machine. Replace '<my-github-repo>' with your actual GitHub repository path. This is the first step in contributing code changes.

```sh
git clone https://github.com/<my-github-repo>/solace-aws-ha-quickstart
```

--------------------------------

### Configure TLS/SSL for Secure Connections

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

Sets up secure TLS connections to the Solace broker, including certificate validation and optional client authentication. Requires trust store and keystore configurations.

```properties
# TLS connection configuration
name=secureSinkConnector
connector.class=com.solace.connector.kafka.connect.sink.SolaceSinkConnector
tasks.max=2
topics=secure-events

# Use secure protocol (tcps://)
sol.host=tcps://secure-broker.example.com:55443
sol.username=tls-user
sol.password=tls-password
sol.vpn_name=secure-vpn
sol.topics=secure/events

# TLS validation settings
sol.ssl_validate_certificate=true
sol.ssl_validate_certicate_date=true

# Trust store configuration (server certificate validation)
sol.ssl_trust_store=/opt/kafka-connect/certs/truststore.jks
sol.ssl_trust_store_password=truststore-secret
sol.ssl_trust_store_format=JKS

# Optional: Restrict accepted common names
sol.ssl_trusted_common_name_list=broker.example.com,*.example.com

# Client certificate authentication (mutual TLS)
sol.authentication_scheme=AUTHENTICATION_SCHEME_CLIENT_CERTIFICATE
sol.ssl_key_store=/opt/kafka-connect/certs/keystore.jks
sol.ssl_key_store_password=keystore-secret
sol.ssl_key_store_format=JKS
sol.ssl_private_key_alias=client-cert
sol.ssl_private_key_password=private-key-secret

# Optional: Exclude older TLS protocols
sol.ssl_excluded_protocols=SSLv3,TLSv1,TLSv1.1

# Record processor
sol.record_processor_class=com.solace.connector.kafka.connect.sink.recordprocessor.SolSimpleRecordProcessor
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
```

--------------------------------

### Kerberos login.conf Configuration

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

Specifies the login module configuration for Kerberos authentication, including keytab and principal details. This file is used by the Java client for authentication.

```ini
SolaceGSS {
    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    keyTab="/opt/kerberos/kafka-connector.keytab"
    storeKey=true
    useTicketCache=false
    principal="kafka-connector@CORP.EXAMPLE.COM"
    debug=false;
};
```

--------------------------------

### Check PubSub+ Sink Connector Status (Shell)

Source: https://github.com/solaceproducts/pubsubplus-connector-kafka-sink/blob/master/README.md

This command retrieves the current status of the PubSub+ Sink Connector running in a distributed Kafka Connect cluster. It queries the Kafka Connect REST API for the status of a specific connector, identified by its name. This is useful for diagnosing startup issues.

```shell
curl 18.218.82.209:8083/connectors/solaceSinkConnector/status | jq
```

--------------------------------

### Configure Message Replay from Kafka Offset

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

Enables the connector to replay messages from a specific Kafka offset, allowing for reprocessing of historical data. Can be set to replay from the beginning or a specific offset.

```properties
# Replay configuration - start from beginning of topic
name=replaySinkConnector
connector.class=com.solace.connector.kafka.connect.sink.SolaceSinkConnector
tasks.max=1
topics=historical-events

sol.host=tcp://broker.example.com:55555
sol.username=replay-user
sol.password=replay-pass
sol.vpn_name=replay-vpn
sol.topics=replay/events

# Replay from the beginning of the Kafka topic (offset 0)
sol.kafka_replay_offset=0

# Or replay from a specific offset
# sol.kafka_replay_offset=1000

# Note: The same offset is applied to all partitions
# Remove this property to resume from last committed offset

sol.record_processor_class=com.solace.connector.kafka.connect.sink.recordprocessor.SolSimpleRecordProcessor
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
```

--------------------------------

### Committing Changes in Git

Source: https://github.com/solaceproducts/pubsubplus-connector-kafka-sink/blob/master/CONTRIBUTING.md

This command stages and commits all modified files with a descriptive message. The '-a' option automatically stages modified files. Use descriptive messages to explain the purpose of your commit.

```sh
git commit -a -m "Your Commit Message"
```

--------------------------------

### Enable DEBUG Logging for Solace JCSMP (Log4j Properties)

Source: https://github.com/solaceproducts/pubsubplus-connector-kafka-sink/blob/master/README.md

This Log4j configuration snippet enables DEBUG level logging specifically for the Solace JCSMP library within Kafka Connect. This is a targeted approach to gather detailed information for troubleshooting PubSub+ connection issues. Remember to revert to INFO or WARN for production environments.

```properties
log4j.logger.com.solacesystems.jcsmp=DEBUG
```

--------------------------------

### Configure Kafka Connect Log4j Appender (Log4j Properties)

Source: https://github.com/solaceproducts/pubsubplus-connector-kafka-sink/blob/master/README.md

This snippet shows a Log4j configuration for the Kafka Connect distributed mode to redirect logs to a file. It defines a rolling file appender named 'file' that writes logs to '/var/log/kafka/connect.log' with a specific pattern and size limits. This is crucial for troubleshooting in distributed environments.

```properties
log4j.rootLogger=INFO, file
log4j.appender.file=org.apache.log4j.RollingFileAppender
log4j.appender.file.File=/var/log/kafka/connect.log
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=[%d] %p %m (%c:%L)%n
log4j.appender.file.MaxFileSize=10MB
log4j.appender.file.MaxBackupIndex=5
log4j.appender.file.append=true
```

--------------------------------

### Configure Solace Kafka Sink Connector (JSON for Distributed Mode)

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

JSON configuration for deploying the sink connector in distributed mode via the Kafka Connect REST API. Enables centralized management and load balancing. Includes settings for connector class, tasks, converters, Kafka topics, Solace connection details, and SSL configuration.

```json
{
  "name": "solaceSinkConnector",
  "config": {
    "connector.class": "com.solace.connector.kafka.connect.sink.SolaceSinkConnector",
    "tasks.max": "3",
    "key.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
    "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
    "group.id": "solSinkConnectorGroup",
    "topics": "orders,inventory,shipments",
    "sol.host": "tcps://broker.example.com:55443",
    "sol.username": "kafka-sink-user",
    "sol.password": "secure-password",
    "sol.vpn_name": "production-vpn",
    "sol.topics": "kafka/orders,kafka/inventory,kafka/shipments",
    "sol.record_processor_class": "com.solace.connector.kafka.connect.sink.recordprocessor.SolSimpleRecordProcessor",
    "sol.use_transactions_for_queue": "true",
    "sol.autoflush.size": "100",
    "sol.ssl_validate_certificate": "true",
    "sol.ssl_trust_store": "/opt/certs/truststore.jks",
    "sol.ssl_trust_store_password": "truststore-pass"
  }
}
```

--------------------------------

### Pushing Git Branch to Origin

Source: https://github.com/solaceproducts/pubsubplus-connector-kafka-sink/blob/master/CONTRIBUTING.md

This command pushes your local branch ('my-fix-branch') to your fork on GitHub (origin). This makes your changes available for creating a pull request.

```sh
git push origin my-fix-branch
```

--------------------------------

### Configure Kafka Sink Connector Properties

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

Sets various properties for the Solace Kafka Sink Connector, including auto-flush size, connection retry parameters, error handling, and the record processor class. These properties control how the connector interacts with Kafka and Solace.

```properties
sol.autoflush.size=100

sol.channel_properties.connect_retries=5
sol.channel_properties.reconnect_retries=10
sol.channel_properties.reconnect_retry_wait_in_millis=5000
sol.channel_properties.keep_alive_interval_in_millis=3000
sol.channel_properties.keep_alive_limit=5

sol.record_processor.error.ignore=false

sol.record_processor_class=com.solace.connector.kafka.connect.sink.recordprocessor.SolSimpleRecordProcessor
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
```

--------------------------------

### Update Connector Configuration using cURL

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

This snippet demonstrates how to update the configuration of the Solace Sink Connector using a cURL command. It specifies the Solace topics to which Kafka messages should be routed and the maximum number of tasks for the connector.

```bash
curl -X PUT -H "Content-Type: application/json" \
  -d '{"sol.topics": "kafka/new-topic", "tasks.max": "5"}' \
  http://localhost:8083/connectors/solaceSinkConnector/config
```

--------------------------------

### Implement Custom Record Processor for Solace Sink Connector

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

This Java code defines a custom record processor that implements the `SolRecordProcessorIF` interface. It allows for transforming Kafka SinkRecords into Solace BytesXMLMessage objects with custom mapping logic, including setting metadata and transforming the payload. The processor can be configured via connector properties.

```java
package com.example.connector;

import com.solace.connector.kafka.connect.sink.SolRecordProcessorIF;
import com.solacesystems.jcsmp.BytesXMLMessage;
import com.solacesystems.jcsmp.JCSMPFactory;
import com.solacesystems.jcsmp.SDTMap;
import org.apache.kafka.connect.sink.SinkRecord;

import java.nio.charset.StandardCharsets;
import java.util.Map;

public class CustomRecordProcessor implements SolRecordProcessorIF {

    private String customPrefix;

    @Override
    public void configure(Map<String, ?> configs) {
        // Access connector configuration during initialization
        this.customPrefix = (String) configs.getOrDefault("custom.prefix", "default");
    }

    @Override
    public BytesXMLMessage processRecord(String skey, SinkRecord record) {
        BytesXMLMessage msg = JCSMPFactory.onlyInstance().createMessage(BytesXMLMessage.class);

        // Set message metadata from Kafka record
        SDTMap userProperties = JCSMPFactory.onlyInstance().createMap();
        try {
            userProperties.putString("k_topic", record.topic());
            userProperties.putInteger("k_partition", record.kafkaPartition());
            userProperties.putLong("k_offset", record.kafkaOffset());
            userProperties.putLong("k_timestamp", record.timestamp());
            userProperties.putString("processor", "CustomRecordProcessor");
        } catch (Exception e) {
            // Handle SDTException
        }
        msg.setProperties(userProperties);
        msg.setApplicationMessageType(customPrefix + ":" + record.topic());

        // Transform and set message payload
        Object value = record.value();
        if (value instanceof byte[]) {
            msg.writeAttachment((byte[]) value);
        } else if (value instanceof String) {
            msg.writeAttachment(((String) value).getBytes(StandardCharsets.UTF_8));
        }

        // Set correlation ID from Kafka key if present
        if (record.key() != null) {
            msg.setCorrelationId(record.key().toString());
        }

        return msg;
    }
}

// Configure in properties file:
// sol.record_processor_class=com.example.connector.CustomRecordProcessor
```

--------------------------------

### Configure Transactional Reliability for Queues/Topics

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

Sets up transactional message delivery to ensure guaranteed reliability when sending messages to Solace PubSub+ queues or topics. Can be configured to use transactions for queues only or both queues and topics.

```properties
# High reliability configuration with transactions
name=reliableSinkConnector
connector.class=com.solace.connector.kafka.connect.sink.SolaceSinkConnector
tasks.max=3
topics=critical-orders

sol.host=tcp://ha-broker.example.com:55555,tcp://dr-broker.example.com:55555
sol.username=reliable-user
sol.password=reliable-pass
sol.vpn_name=production

# Send to a durable queue for guaranteed delivery
sol.queue=orders-queue

# Transaction settings
sol.use_transactions_for_queue=true
sol.use_transactions_for_topics=false
```

--------------------------------

### Creating a New Git Branch

Source: https://github.com/solaceproducts/pubsubplus-connector-kafka-sink/blob/master/CONTRIBUTING.md

This command creates a new branch named 'my-fix-branch' based on the 'master' branch. It's essential to make changes on a separate branch to keep the main codebase clean and organized.

```sh
git checkout -b my-fix-branch master
```

--------------------------------

### Rebasing Git Branch with Upstream

Source: https://github.com/solaceproducts/pubsubplus-connector-kafka-sink/blob/master/CONTRIBUTING.md

This sequence of commands fetches the latest changes from the 'upstream' repository and rebases your local branch onto the 'master' branch. This ensures your changes are based on the most recent code. It's recommended before pushing if the branch hasn't been pushed yet.

```sh
git remote add upstream https://github.com/SolaceLabs/solace-kafka-connector-sink
git fetch upstream
git rebase upstream/master
```

--------------------------------

### Merging Git Branch

Source: https://github.com/solaceproducts/pubsubplus-connector-kafka-sink/blob/master/CONTRIBUTING.md

These commands merge the 'master' branch into your current branch and then merge your feature branch ('my-fix-branch') into 'master'. This is used to synchronize your work with the main repository after your branch has been pushed to origin.

```sh
git merge master
git checkout master
git merge my-fix-branch
```

--------------------------------

### Enable Kafka Headers to Solace User Properties Mapping

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

Configures the Solace Kafka Sink Connector to automatically copy Kafka record headers to Solace message user properties. This is useful for preserving metadata during event streaming. Note that if multiple values exist for the same header key, only the last one is retained.

```properties
name=headersSinkConnector
connector.class=com.solace.connector.kafka.connect.sink.SolaceSinkConnector
tasks.max=1
topics=events-with-headers

sol.host=tcp://broker.example.com:55555
sol.username=headers-user
sol.password=headers-pass
sol.vpn_name=default
sol.topics=events/forwarded

# Enable automatic Kafka headers to Solace user properties mapping
sol.emit.kafka.record.headers.enabled=true

sol.record_processor_class=com.solace.connector.kafka.connect.sink.recordprocessor.SolSimpleRecordProcessor
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
```

--------------------------------

### SolSimpleRecordProcessor: Basic Kafka to Solace Message Conversion (Java)

Source: https://context7.com/solaceproducts/pubsubplus-connector-kafka-sink/llms.txt

The default record processor that converts Kafka records to Solace messages with a binary payload. It preserves Kafka metadata by setting Kafka topic, partition, and offset as user properties on the Solace message. The ApplicationMessageType is set to 'ResendOfKafkaTopic: <topic>', and the record value becomes the binary attachment.

```properties
name=simpleSinkConnector
connector.class=com.solace.connector.kafka.connect.sink.SolaceSinkConnector
tasks.max=1
topics=orders
sol.host=tcp://localhost:55555
sol.username=default
sol.password=default
sol.vpn_name=default
sol.topics=solace/orders
sol.record_processor_class=com.solace.connector.kafka.connect.sink.recordprocessor.SolSimpleRecordProcessor
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
key.converter=org.apache.kafka.connect.storage.StringConverter
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.