Introduction

Amazon MSK supports SASL/SCRAM for username-and-password authentication, which is a common choice when Kafka clients need authenticated and encrypted access without using IAM auth. The Java side is straightforward once the cluster, secret, and network path are ready, but small configuration mistakes usually cause the connection to fail immediately.

What Must Exist Before Java Can Connect

Before writing client code, verify four things in AWS:

1. SASL/SCRAM is enabled on the MSK cluster.
2. A Secrets Manager secret is associated with the cluster.
3. The client can reach the brokers over the correct network path.
4. You retrieved the BootstrapBrokerStringSaslScram value for that cluster.

AWS exposes the correct broker list through get-bootstrap-brokers. Use the SASL/SCRAM-specific broker string, not the plaintext or IAM value.

Required Kafka Client Properties

For MSK with SASL/SCRAM, the core client settings are security.protocol=SASL_SSL, the appropriate SCRAM mechanism, and a JAAS configuration containing the username and password.

1package com.example.msk;
2
3import java.util.Properties;
4
5public class MskClientConfig {
6    public static Properties baseProperties() {
7        Properties props = new Properties();
8        props.put("bootstrap.servers",
9                "b-1.example.kafka.us-east-1.amazonaws.com:9096,"
10              + "b-2.example.kafka.us-east-1.amazonaws.com:9096");
11        props.put("security.protocol", "SASL_SSL");
12        props.put("sasl.mechanism", "SCRAM-SHA-512");
13        props.put(
14                "sasl.jaas.config",
15                "org.apache.kafka.common.security.scram.ScramLoginModule required "
16              + "username=\"alice\" password=\"secret-password\";"
17        );
18        return props;
19    }
20}

If your cluster is configured for SCRAM-SHA-256, change the mechanism accordingly. The rest of the setup is the same.

Producer Example

Once the properties are correct, producing is no different from any other Kafka Java client.

1package com.example.msk;
2
3import java.util.Properties;
4import org.apache.kafka.clients.producer.KafkaProducer;
5import org.apache.kafka.clients.producer.ProducerRecord;
6import org.apache.kafka.common.serialization.StringSerializer;
7
8public class MskProducerDemo {
9    public static void main(String[] args) {
10        Properties props = MskClientConfig.baseProperties();
11        props.put("key.serializer", StringSerializer.class.getName());
12        props.put("value.serializer", StringSerializer.class.getName());
13
14        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
15            producer.send(new ProducerRecord<>("orders", "order-1", "created")).get();
16            System.out.println("Message sent");
17        } catch (Exception ex) {
18            ex.printStackTrace();
19        }
20    }
21}

This code is runnable if the broker list, topic, and credentials are valid.

Consumer Example

Consumers use the same security properties plus deserializers and a group id.

1package com.example.msk;
2
3import java.time.Duration;
4import java.util.Collections;
5import java.util.Properties;
6import org.apache.kafka.clients.consumer.ConsumerRecord;
7import org.apache.kafka.clients.consumer.ConsumerRecords;
8import org.apache.kafka.clients.consumer.KafkaConsumer;
9import org.apache.kafka.common.serialization.StringDeserializer;
10
11public class MskConsumerDemo {
12    public static void main(String[] args) {
13        Properties props = MskClientConfig.baseProperties();
14        props.put("group.id", "orders-demo");
15        props.put("auto.offset.reset", "earliest");
16        props.put("key.deserializer", StringDeserializer.class.getName());
17        props.put("value.deserializer", StringDeserializer.class.getName());
18
19        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
20            consumer.subscribe(Collections.singletonList("orders"));
21            ConsumerRecords<String, String> records =
22                    consumer.poll(Duration.ofSeconds(5));
23
24            for (ConsumerRecord<String, String> record : records) {
25                System.out.printf(
26                        "offset=%d key=%s value=%s%n",
27                        record.offset(), record.key(), record.value());
28            }
29        }
30    }
31}

Where Failures Usually Come From

When this setup fails, the problem is often outside the Java code. If the client is not inside the right VPC or cannot reach the brokers through peering, VPN, Transit Gateway, or public access, authentication never gets a chance to succeed.

Another frequent issue is choosing the wrong bootstrap broker string. MSK publishes different values for TLS, IAM, and SASL/SCRAM clients. The names are similar enough that copy-paste mistakes are common.

Certificate trust can also break connections. Because SASL/SCRAM on MSK uses TLS, your Java runtime must trust the broker certificate chain. In most standard environments the default trust store is enough, but locked-down corporate runtimes sometimes require extra work.

Handling Secrets More Safely

Hardcoding the username and password is acceptable for a minimal example but not for production. A better pattern is to fetch credentials from environment variables or a secret manager and build the JAAS config string at startup.

1String username = System.getenv("MSK_USERNAME");
2String password = System.getenv("MSK_PASSWORD");
3props.put(
4        "sasl.jaas.config",
5        "org.apache.kafka.common.security.scram.ScramLoginModule required "
6      + "username=\"" + username + "\" password=\"" + password + "\";"
7);

That keeps credentials out of source control while still using the standard Kafka client.

Common Pitfalls

The biggest pitfall is mixing authentication modes. IAM examples use callback handlers and different mechanisms, but those settings do not belong in a SASL/SCRAM client.

Another common error is using PLAINTEXT or SASL_PLAINTEXT instead of SASL_SSL. MSK's SASL/SCRAM setup expects TLS encryption with the authentication flow.

Developers also forget that broker connectivity is an infrastructure concern. Security groups, route tables, DNS resolution, and cluster access settings are just as important as the Java properties file.

Finally, avoid pasting credentials into exception logs or debugging output. SASL configuration strings often contain the password in plain text.

Summary

• Use the MSK broker list returned for BootstrapBrokerStringSaslScram.
• Configure Java clients with security.protocol=SASL_SSL and the matching SCRAM mechanism.
• Producer and consumer code look normal once the security properties are correct.
• Most failures come from network access, wrong broker strings, or bad credentials.
• Keep usernames and passwords out of source code in production systems.