Need Help Creating Scram-SHA-512 User and Setting ECLs in Kafka via Strimzi API

[rizsk1220]

Hi everyone,
I have been working on Strimzi API development, and I successfully created Kafka topics using the provided Maven dependencies. Currently, I am working on creating a Kafka user with Scram-SHA-512 authentication, but I am struggling with the following tasks:

1. How to programmatically create a Scram-SHA-512 user using Strimzi API.
2. How to set the user secret and configure the ACLs.
   I’ve been unable to find concrete examples or documentation for this specific use case.
   If anyone has any working code examples, documentation, or guidance, I would greatly appreciate it.
   Thanks in advance!

Topic working code:
try (KubernetesClient client = new KubernetesClientBuilder().build()) {
KafkaTopic topic = new KafkaTopicBuilder()
.withApiVersion("kafka.strimzi.io/v1beta2\n" +
"kind: KafkaTopic")
.withNewMetadata()
.withName(TOPIC_NAME)
.withNamespace(NAMESPACE)
.withLabels(Map.of("strimzi.io/cluster", KAFKA_NAME))
.endMetadata()
.withNewSpec()
.withPartitions(3)
.withReplicas(3)
.withConfig(Map.of("retention.ms", 7200000,
"segment.bytes", 107374182,
"min.insync.replicas", 2))
.endSpec()
.build();

        LOGGER.info("Creating the topic");
        Crds.topicOperation(client).inNamespace(NAMESPACE).resource(topic).create();

        LOGGER.info("Waiting for the topic to be ready");
        Crds.topicOperation(client).inNamespace(NAMESPACE).withName(TOPIC_NAME).waitUntilCondition(KafkaTopic.isReady(), 5, TimeUnit.MINUTES);

        LOGGER.info("Topic is ready");

User code which is not working

String encodedPassword = Base64.getEncoder().encodeToString(password.getBytes());
try (KubernetesClient client = new KubernetesClientBuilder().build()) {

        // Step 1: Create a Secret for the password
        Map<String, String> secretData = new HashMap<>();
        secretData.put("password", encodedPassword);
        Secret passwordSecret = new SecretBuilder()
                .withMetadata(new ObjectMetaBuilder()
                        .withName(kafkaUserName + "-password")
                        .withNamespace(namespace)
                        .build())
                .withType("Opaque")
                .withData(secretData)
                .build();
        // Create or replace the Secret using the modern method
        Secret secret = client.secrets().inNamespace(namespace).resource(passwordSecret).create();


        System.out.println("Password Secret created successfully.");

        // Step 2: Create the KafkaUser resource
        KafkaUserScramSha512ClientAuthenticationBuilder authentication = new KafkaUserScramSha512ClientAuthenticationBuilder();
        authentication.withPassword() ///. NEED Help here



        KafkaUserAuthorizationSimpleBuilder authorizationBuilder = new KafkaUserAuthorizationSimpleBuilder()
                .addNewAcl()
                .withResource()   // NEED help here
                .withType("topic")
                .withName("my-topic")
                .withPatternType("literal")
                .endResource()
                .withOperation("Read")
                .endAcl();

        KafkaUser user = new KafkaUserBuilder()         /// Need help here
                .withApiVersion("kafka.strimzi.io/v1beta2\n" +
                        "kind: KafkaUser")
                .withNewMetadata()
                    .withName(kafkaUserName)
                     .withNamespace(NAMESPACE)
                .withLabels(Map.of("strimzi.io/cluster", KAFKA_NAME))
                .endMetadata()
                .withNewSpec()
                .withAuthentication(authentication)
                .withAuthorization(authorizationBuilder.build())
                .endSpec()
                .build();

[scholzj]

First of all, you should not have these parts there:

                .withApiVersion("kafka.strimzi.io/v1beta2\n" +
                        "kind: KafkaUser")

Just set the Spec and the Metadata.

For the rest, just use some IDE and follow the DSL language it offers you. You should end up for example with something like this:

            KafkaUser user = new KafkaUserBuilder()
                    .withNewMetadata()
                        .withName(kafkaUserName)
                        .withNamespace(NAMESPACE)
                        .withLabels(Map.of("strimzi.io/cluster", KAFKA_NAME))
                    .endMetadata()
                    .withNewSpec()
                        .withNewKafkaUserScramSha512ClientAuthentication()
                            .withNewPassword()
                                .withNewValueFrom()
                                    .withNewSecretKeyRef("my-secret", "my-key", false)
                                .endValueFrom()
                            .endPassword()
                        .endKafkaUserScramSha512ClientAuthentication()
                    .endSpec()
                    .build();

[rizsk1220]

how to add acl to the above code

[scholzj]

What is the code you tried?

[rizsk1220]

KafkaUserAuthorizationSimpleBuilder authorizationBuilder = new KafkaUserAuthorizationSimpleBuilder()
.addNewAcl()
.withResource(). --- getting error here
.withType("topic")
.withName("my-topic")
.withPatternType("literal")
.endResource()
.withOperation("Read")
.endAcl();

[scholzj]

You have to follow the method signatures. If you use withResource(...), you have to provide the resource as parameter. As I suggested, if you use some IDE, the autocompletion should help with things like this.

[rizsk1220]

let me check

[rizsk1220]

how can i add secret in K8 and how to use custom password to the above code

[scholzj]

That is what the example I gave you does.

[rizsk1220]

ok i will try and let you know if i face any issues,can you also help me in setting the acl too

[rizsk1220]

can you explain this line
.withNewSecretKeyRef("my-secret", "my-key", false)

what does falsee means here

[scholzj]

If you follow the API reference - https://strimzi.io/docs/operators/latest/full/configuring.html - you can go through the schema and see the meaning of the different fields.

[rizsk1220]

Final code

mport io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientBuilder;
import io.strimzi.api.kafka.Crds;
import io.strimzi.api.kafka.model.user.KafkaUser;
import io.strimzi.api.kafka.model.user.KafkaUserBuilder;
import io.strimzi.api.kafka.model.user.acl.AclOperation;
import io.strimzi.api.kafka.model.user.acl.AclResourcePatternType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class CreateStrimziUser {
private static final Logger LOGGER = LoggerFactory.getLogger(CreateStrimziUser.class);
private static final String NAMESPACE = "digital-strimzi-kafka";
private static final String KAFKA_NAME = "digital-strimzi";
private static final String TOPIC_NAME = "my-topic-strimzi";
private static final String kafkaUserName = "my-topic-user";

public static void main(String[] args) {
    try (KubernetesClient client = new KubernetesClientBuilder().build()) {
        KafkaUser user = new KafkaUserBuilder()
                .withNewMetadata()
                .withName(kafkaUserName)
                .withNamespace(NAMESPACE)
                .withLabels(Map.of("strimzi.io/cluster", KAFKA_NAME))
                .endMetadata()
                .withNewSpec()
                .withNewKafkaUserScramSha512ClientAuthentication()
                .withNewPassword()
                .withNewValueFrom()
                .withNewSecretKeyRef("my-secret", "my-key", false)
                .endValueFrom()
                .endPassword()
                .endKafkaUserScramSha512ClientAuthentication()
                .withNewKafkaUserAuthorizationSimple()
                .addNewAcl()
                .withNewAclRuleTopicResource()
                .withName(TOPIC_NAME)
                .withPatternType(AclResourcePatternType.valueOf("prefix"))
                .endAclRuleTopicResource()
                .withOperation(AclOperation.valueOf("All"))
                .endAcl()
                .endKafkaUserAuthorizationSimple()
                .endSpec()
                .build();

        LOGGER.info("Creating the user");
        Crds.kafkaUserOperation(client).inNamespace(NAMESPACE).resource(user).create();

        LOGGER.info("Waiting for the user to be ready");
        Crds.kafkaUserOperation(client).inNamespace(NAMESPACE).withName(kafkaUserName)
                .waitUntilCondition(KafkaUser.isReady(), 5, TimeUnit.MINUTES);

        LOGGER.info("User is ready");

    }
}

}

Error:

Exception in thread "main" java.lang.IllegalArgumentException: No enum constant io.strimzi.api.kafka.model.user.acl.AclResourcePatternType.prefix
at java.base/java.lang.Enum.valueOf(Enum.java:273)
at io.strimzi.api.kafka.model.user.acl.AclResourcePatternType.valueOf(AclResourcePatternType.java:10)
at org.example.CreateStrimziUser.main(CreateStrimziUser.java:45)

[rizsk1220]

corrected the code

package org.example;

import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientBuilder;
import io.strimzi.api.kafka.Crds;
import io.strimzi.api.kafka.model.user.KafkaUser;
import io.strimzi.api.kafka.model.user.KafkaUserBuilder;
import io.strimzi.api.kafka.model.user.acl.AclOperation;
import io.strimzi.api.kafka.model.user.acl.AclResourcePatternType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class CreateStrimziUser {
private static final Logger LOGGER = LoggerFactory.getLogger(CreateStrimziUser.class);
private static final String NAMESPACE = "digital-strimzi-kafka";
private static final String KAFKA_NAME = "digital-strimzi";
private static final String TOPIC_NAME = "my-topic-strimzi";
private static final String kafkaUserName = "my-topic-user";

public static void main(String[] args) {
    try (KubernetesClient client = new KubernetesClientBuilder().build()) {
        KafkaUser user = new KafkaUserBuilder()
                .withNewMetadata()
                .withName(kafkaUserName)
                .withNamespace(NAMESPACE)
                .withLabels(Map.of("strimzi.io/cluster", KAFKA_NAME))
                .endMetadata()
                .withNewSpec()
                .withNewKafkaUserScramSha512ClientAuthentication()
                .withNewPassword()
                .withNewValueFrom()
                .withNewSecretKeyRef("my-secret", "my-key", false)
                .endValueFrom()
                .endPassword()
                .endKafkaUserScramSha512ClientAuthentication()
                .withNewKafkaUserAuthorizationSimple()
                .addNewAcl()
                .withNewAclRuleTopicResource()
                .withName(TOPIC_NAME)
                .withPatternType(AclResourcePatternType.PREFIX)
                .endAclRuleTopicResource()
                .withOperations(AclOperation.ALL)
                .endAcl()
                .endKafkaUserAuthorizationSimple()
                .endSpec()
                .build();

        LOGGER.info("Creating the user");
        Crds.kafkaUserOperation(client).inNamespace(NAMESPACE).resource(user).create();

        LOGGER.info("Waiting for the user to be ready");
        Crds.kafkaUserOperation(client).inNamespace(NAMESPACE).withName(kafkaUserName)
                .waitUntilCondition(KafkaUser.isReady(), 5, TimeUnit.MINUTES);

        LOGGER.info("User is ready");

    }
}

}

Getting Error
SLF4J: Actual binding is of type [org.slf4j.impl.SimpleLoggerFactory]
[main] INFO org.example.CreateStrimziUser - Creating the user
[main] WARN io.fabric8.kubernetes.client.dsl.internal.VersionUsageUtils - The client is using resource type 'kafkausers' with unstable version 'v1beta2'
[main] INFO org.example.CreateStrimziUser - Waiting for the user to be ready
Exception in thread "main" io.fabric8.kubernetes.client.KubernetesClientTimeoutException: Timed out waiting for [300000] milliseconds for [KafkaUser] with name:[my-topic-user] in namespace [digital-strimzi-kafka].
at io.fabric8.kubernetes.client.dsl.internal.BaseOperation.waitUntilCondition(BaseOperation.java:946)
at io.fabric8.kubernetes.client.dsl.internal.BaseOperation.waitUntilCondition(BaseOperation.java:98)
at org.example.CreateStrimziUser.main(CreateStrimziUser.java:55)

[scholzj]

You would need to check the operator logs etc. to see why it is not ready.

[rizsk1220]

Thank You, I'm able to solve it what if i want to update the acl to the existing user, i dont see replace method. Do i need to use update() or edit the specification.

Working code i have as of now

List aclRules = new ArrayList<>();
aclRules.add(new AclRuleBuilder()
.withNewAclRuleTopicResource()
.withName("conditional-dynamic-config-topic-2")
.withPatternType(AclResourcePatternType.PREFIX)
.endAclRuleTopicResource()
.withOperations(AclOperation.WRITE)
.build());

        aclRules.add(new AclRuleBuilder()
                .withNewAclRuleTopicResource()
                .withName("conditional-dynamic-config-topic-10")
                .withPatternType(AclResourcePatternType.LITERAL)
                .endAclRuleTopicResource()
                .withOperations(AclOperation.READ,AclOperation.WRITE)
                .build());
        aclRules.add(new AclRuleBuilder()
                .withNewAclRuleGroupResource()
                .withName("conditional-dynamic-group")
                .withPatternType(AclResourcePatternType.PREFIX)
                .endAclRuleGroupResource()
                .withOperations(AclOperation.READ)
                .build());
        KafkaUser user;
        user = new KafkaUserBuilder()
                .withNewMetadata()
                .withName(kafkaUserName)
                .withNamespace(NAMESPACE)
                .withLabels(Map.of("strimzi.io/cluster", KAFKA_NAME))
                .endMetadata()
                .withNewSpec()
                .withNewKafkaUserScramSha512ClientAuthentication()
                .withNewPassword()
                .withNewValueFrom()
                .withNewSecretKeyRef("password", kafkaUserName , false)
                .endValueFrom()
                .endPassword()
                .endKafkaUserScramSha512ClientAuthentication()
                .withNewKafkaUserAuthorizationSimple()
                .withAcls(aclRules)
                .endKafkaUserAuthorizationSimple()
                .endSpec()
                .build();

        LOGGER.info("Creating the user");
        Crds.kafkaUserOperation(client).inNamespace(NAMESPACE).resource(user).create();

        LOGGER.info("Waiting for the user to be ready");
        Crds.kafkaUserOperation(client).inNamespace(NAMESPACE).withName(kafkaUserName)
                .waitUntilCondition(KafkaUser.isReady(), 5, TimeUnit.MINUTES);

        LOGGER.info("User is ready");

[scholzj]

I think there is some patch() method or something like that in Fabric8.

[rizsk1220]

Worked method patch for update and delete to delete kafka topics and users.
is there any method to list the topics and user as well, i dont see anything such.

Crds.topicOperation(client).inNamespace(NAMESPACE).resource(topic).create(); something like this to list.

[scholzj]

You should really:

• Check the Fabric8 Kubernetes client docs to see what the different methods are
• Get some IDE with autocompletion as that lets you figure out everything

It is not possible to give you detailed guidance for every single command.

[rizsk1220]

i figured it out and its working. Thank you!