Kafka Plugin: support for SASL/SCRAM mechanisms

[franky-m]

Description

Add support for SASL/SCRAM mechanisms

example config:

authentication:
  sasl:
    scram:
      username: user1234
      password: password1234
      mechanism: SCRAM-SHA-512

Issues Resolved

Resolves #4241

Check List

• New functionality includes testing.
• New functionality has a documentation issue. Please link to it in this PR.
  • New functionality has javadoc added
• Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[chenqi0805]

@franky-m Thanks for adding this enhancement! Could you rebase your change off the current main branch?

[franky-m]

@chenqi0805 I did the rebase. Happy to help.

[chenqi0805]

@franky-m Thanks a lot for the contribution! This is a significant enhancement and your change mostly look good to me. One thing to point out is we do support credential refreshment from aws secrets through sasl, which would require incorporate your ScramAuthConfig into

data-prepper/data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/authenticator/DynamicBasicCredentialsProvider.java

Lines 33 to 50 in 76d9640

	public void refresh(final KafkaConnectionConfig newConfig) {
	final AuthConfig authConfig = newConfig.getAuthConfig();
	if (Objects.nonNull(authConfig)) {
	AuthConfig.SaslAuthConfig saslAuthConfig = authConfig.getSaslAuthConfig();
	if (Objects.nonNull(saslAuthConfig) && Objects.nonNull(saslAuthConfig.getPlainTextAuthConfig())) {
	final PlainTextAuthConfig plainTextAuthConfig = newConfig.getAuthConfig().getSaslAuthConfig()
	.getPlainTextAuthConfig();
	final String newUsername = plainTextAuthConfig.getUsername();
	final String newPassword = plainTextAuthConfig.getPassword();
	readWriteLock.writeLock().lock();
	try {
	basicCredentials = new BasicCredentials(newUsername, newPassword);
	} finally {
	readWriteLock.writeLock().unlock();
	}
	}
	}
	}

since it is desirable to support refreshing SASL/SCRAM credentials just as SASL/PLAIN. This can come in a separate PR (We should create an issue if you want to merge this PR first) or in the current PR.

[franky-m]

Hi @chenqi0805 although it would be a good idea to implement the credential refreshment right away I might have to pass on this one. I tried to comprehend the architecture behind it and could not really grasp it. Thus I have no understanding on where to put what to make this a clean and complete implementation. So I would prefer to keep this out of this particular PR.

[dlvenable]

@franky-m,

Thank you for this great contribution. I'm happy to accept this without the credentials refreshment.

I do have one question. And we also need to wait for the tests to pass before merging.

[dlvenable]

What are the valid values here?

[franky-m]

Kafka excepts SCRAM-SHA-256 and SCRAM-SHA-512 as mechanisms. Do you think it would be necessary to validate at this point? You will find out pretty much immediately if you enter something invalid so I thought it would be a bit overkill.

[dlvenable]

Thanks @franky-m . Adding validation would help. If you'd like to do this in a follow-on PR that would help.

[dlvenable]

Very nice! Thank you for adding these integration tests.

[chenqi0805]

Hi @chenqi0805 although it would be a good idea to implement the credential refreshment right away I might have to pass on this one. I tried to comprehend the architecture behind it and could not really grasp it. Thus I have no understanding on where to put what to make this a clean and complete implementation. So I would prefer to keep this out of this particular PR.

@franky-m No problem! I can create an issue for that and take over.

[dlvenable]

Thank you @franky-m for this contribution!