Kafka source: support SASL/SCRAM mechanisms

[franky-m]

Is your feature request related to a problem? Please describe. Currently, Kafka source only supports SASL/PLAIN authentication mechanism, but apparently no SASL/SCRAM-SHA-256 and SASL/SCRAM-SHA-512.

Describe the solution you'd like

Extend Data Prepper’s authentication options to include mechanisms such as SCRAM-SHA-512.

Example:

pipeline:
  name: kafka-pipeline
  source:
    kafka:
      bootstrap_servers:
        - 127.0.0.1:9093
      topics:
        - name: topic1
          group_id: groupID1
      authentication:
        sasl:
          plaintext:
            mechanism: SCRAM-SHA-512 # or SCRAM-SHA-256 or plain
            username: your_kafka_username
            password: your_kafka_password

Additional context

Many Kafka deployments rely on SCRAM mechanisms for improved security. Users who require SCRAM-SHA-512 authentication need this feature to seamlessly integrate Data Prepper into their existing Kafka infrastructure.

[dlvenable]

@franky-m , This is a great idea. Would you be able to contribute a PR to help with this? We could give you some pointers in the code.

[franky-m]

Hi could you give me the pointers you mentioned? I would try to implement the SASL/SCRAM support myself and if I succeed I would open a PR

[burandobata]

+1

[franky-m]

Hi @dlvenable! Did you have time to collect the pointers you mentioned?

[dlvenable]

@franky-m ,

Yes, I have some references.

First, I think that SASL/SCRAM is different from SASL/PLAIN. So the configuration should probably be a little different in the YAML.

It should have the following structure instead.

authentication:
  sasl:
    scram:
      username: your_kafka_username
      password: your_kafka_password

You can see where we add the current configuraiton in this block:

https://github.com/opensearch-project/data-prepper/blob/64445fac5c42a801bf62b94e0a002353c4065972/data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/AuthConfig.java#L21-L24

You can add something like the following below there to add the scram option.

        @JsonProperty("scram")
        private ScramAuthConfig scramAuthConfig;

Here is code where we set the plain configuration into the Kafka properties:

https://github.com/opensearch-project/data-prepper/blob/af7d1b54561a52ac13e7ba04d2b73a77cb0dc5cb/data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaSecurityConfigurer.java#L120-L132

And this is the code where we call it.

https://github.com/opensearch-project/data-prepper/blob/af7d1b54561a52ac13e7ba04d2b73a77cb0dc5cb/data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaSecurityConfigurer.java#L341-L342

You could add a new condition that would look somewhat like:

...
} else if(Objects.nonNull(saslAuthConfig.getScramAuthConfig())) {
  setScramAuthProperties(properties, saslAuthConfig.getScramAuthConfig());  // new method; maybe it needs the encryption config too
} else ...

[dlvenable]

@franky-m , We did make some changes to support dynamically updating the password if it changes in the underlying source (e.g. AWS Secrets Manager). It isn't necessary to have that implemented, but would be nice.

@chenqi0805 , Can you provide any guidance on how that would be implemented?