Refactor Client-Example ConsumerConfigs #91

OwenCorrigan76 · 2022-12-06T16:18:10Z

Refactor the Producer, Consumer and Streams Config files in the Java client by reducing the complexity and size of the class. Standardise the naming scheme of environment variables used to configure the clients / standardise the envVars used to configure the Kafka client properties as outlined here [1].

Example of standardised Environmental Variables:

KAFKA_<NAME_OF_KAFKA_PROPERTY_DELIMITED_BY_UNDERSCORES>

[1] https://github.com/kyguy/proposals/blob/refactor-client-examples/040-refactor-client-examples.md

scholzj

I know this is just a draft, so not sure what else do you plan. But I think you should go deeper ... I do not think there is any need to have some special treatment for the TLS certificates or OAuth authentication. I think they can be configured from the KAFKA_... env vars as well.

kyguy · 2022-12-06T23:22:10Z

java/kafka/producer/src/main/java/KafkaProducerConfig.java

+                ", oauthTokenEndpointUri='" + oauthTokenEndpointUri + '\'' +
+                ", additionalConfig='" + additionalConfig + '\'' +
+                ", tracingSystem='" + tracingSystem + '\'' +
+                '}';
    }
 }


Undo changes

kyguy · 2022-12-06T23:22:19Z

java/kafka/streams/src/main/java/KafkaStreamsExample.java

@@ -46,8 +46,7 @@ public static void main(String[] args) {
                KafkaClientSupplier supplier = new TracingKafkaClientSupplier();
                streams = new KafkaStreams(builder.build(), props, supplier);
            } else {
-                log.error("Error: TRACING_SYSTEM {} is not recognized or supported!", config.getTracingSystem());
-                streams = new KafkaStreams(builder.build(), props);
+                throw new RuntimeException("Error: TRACING_SYSTEM " + tracingSystem + " is not recognized or supported!");
            }
        } else {
            streams = new KafkaStreams(builder.build(), props);


undo changes

kyguy · 2022-12-06T23:22:35Z

java/kafka/deployment-tracing-jaeger.yaml

-              value: jaeger
---
-apiVersion: apps/v1
-kind: Deployment


Undo changes

Undone (should be correct now)

kyguy · 2022-12-06T23:23:16Z

java/kafka/consumer/src/main/java/KafkaConsumerConfig.java

        }

        if ((config.getOauthAccessToken() != null)
                || (config.getOauthTokenEndpointUri() != null && config.getOauthClientId() != null && config.getOauthRefreshToken() != null)
-                || (config.getOauthTokenEndpointUri() != null && config.getOauthClientId() != null && config.getOauthClientSecret() != null))    {


undo indentation change

Full block of code removed. All security is being handled by the user via the envVars.

kyguy · 2022-12-06T23:24:23Z

README.md

+* `STRIMZI_TOPIC` - name of topic which consumer subscribes  
+* `KAFKA_GROUP_ID` - specifies the consumer group id for the consumer
+* `STRIMZI_MESSAGE_COUNT` - the number of messages the consumer should receive
+* `STRIMZI_LOG_LEVEL` - logging level  
 * `ADDITIONAL_CONFIG` - additional configuration for a consumer application. Notice, that you can also override any previously set variable by setting this. The form is `key=value` records separated by new line character


This can be removed

Including the STRIMZI_ prefixed ones?

kyguy · 2022-12-06T23:27:48Z

README.md

-* `LOG_LEVEL` - logging level  
+* `KAFKA_BOOTSTRAP_SERVERS` - comma-separated host and port pairs that is a list of Kafka broker addresses. The form of pair is `host:port`, e.g. `my-cluster-kafka-bootstrap:9092` 
+* `STRIMZI_TOPIC` - name of topic which consumer subscribes  
+* `KAFKA_GROUP_ID` - specifies the consumer group id for the consumer


Env vars used for Kafka client configuration can be removed.

I would also add a note to the README.md explaining that all Kafka client properties can be configured by adding an env var with the Kafka property name in a capital letters, delimited by "_", and prefixed with "KAFKA"

OwenCorrigan76 · 2022-12-07T14:52:07Z

@scholzj When you say "special treatment", do you mean these lines [1] [2]? I kept these in to give the same behaviour as the code before it. Are you saying we can strip this code out completely? If so, are we removing some of the assistance we're offering to the user? So, the user will have to make sure Ssl. Oauth etc. are configured properly from, their end? Thanks

[1]

client-examples/java/kafka/consumer/src/main/java/KafkaConsumerConfig.java

Lines 90 to 99 in e40023a

    
           if (USER_CONFIGS.containsKey(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG)) { 
        
               log.info("Configuring truststore"); 
        
               props.putAll(DEFAULT_TRUSTSTORE_CONFIGS); 
        
           } 
        
           if (USER_CONFIGS.containsKey(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG) 
        
                   && USER_CONFIGS.containsKey(SslConfigs.SSL_KEYSTORE_KEY_CONFIG)) { 
        
               log.info("Configuring keystore"); 
        
               props.putAll(DEFAULT_KEYSTORE_CONFIGS); 
        
           }

[2]

client-examples/java/kafka/consumer/src/main/java/KafkaConsumerConfig.java

Lines 101 to 111 in e40023a

    
           if ((config.getOauthAccessToken() != null) 
        
                   || (config.getOauthTokenEndpointUri() != null && config.getOauthClientId() != null && config.getOauthRefreshToken() != null) 
        
                   || (config.getOauthTokenEndpointUri() != null && config.getOauthClientId() != null && config.getOauthClientSecret() != null)) { 
        
               log.info("Configuring OAuth"); 
        
               props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;"); 
        
               props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL".equals(props.getProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG)) ? "SASL_SSL" : "SASL_PLAINTEXT"); 
        
               props.put(SaslConfigs.SASL_MECHANISM, "OAUTHBEARER"); 
        
               if (!(USER_CONFIGS.containsKey(SaslConfigs.SASL_MECHANISM) && USER_CONFIGS.get(SaslConfigs.SASL_MECHANISM).equals("PLAIN"))) { 
        
                   props.put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS, config.saslLoginCallbackClass); 
        
               } 
        
           }

scholzj · 2022-12-07T17:55:11Z

Yes, that is the code I mean. Also the (de)serializers for example. You can call it an assistance. But it also makes it unclear how exactly it is configured. Because one day when they need to write their own app, they will need to dig into the code to understand how to actually do it. I think anything what ends up in the consumer properties file should go directly through the KAFKA_ environment variables. The only part which should remain are the once which don't end up there ... e.g. number of messages, tracer initiliazation etc.

OwenCorrigan76 · 2022-12-08T13:07:45Z

@scholzj @kyguy I've removed the Ssl and oAuth configs and only left the USER_CONFIGS in. These are the ones that will have a STRIMZI_ prefix. If you guys give me the go ahead that this is the way to go, I'll address the other comments @kyguy left and then move on to the Producer and Streams. Thanks guys.

scholzj

Some more comments, but it starts looking better. You should also update all the YAML example files to make it work with the new code.

scholzj · 2022-12-08T14:18:37Z

README.md

+* `STRIMZI_TOPIC` - name of topic which consumer subscribes  
+* `KAFKA_GROUP_ID` - specifies the consumer group id for the consumer
+* `STRIMZI_MESSAGE_COUNT` - the number of messages the consumer should receive
+* `STRIMZI_LOG_LEVEL` - logging level  
 * `ADDITIONAL_CONFIG` - additional configuration for a consumer application. Notice, that you can also override any previously set variable by setting this. The form is `key=value` records separated by new line character


You can remove this as well I guess.

@scholzj I've removed ADDITIONAL_CONFIG as it's no longer part of ConsumerConfig. Do you suggest removing the STRIMZI_ prefixed envVar examples too and just leave in the explanation at the beginning of the ##Configuration paragraph?

.yaml files updated with correct prefixes.

scholzj · 2022-12-08T14:20:20Z

java/kafka/consumer/src/main/java/KafkaConsumerConfig.java

+                ", enableAutoCommit='" + enableAutoCommit + '\'' +
+                ", messageCount=" + messageCount +
+                ", tracingSystem='" + tracingSystem + '\'' +
+                kafkaConfigOptionsToString() +


Is this needed? The Kafka client prints its config at startup. If you want to keep this, you should check the formatting to fit into the other fields.

Not needed now as we have kafkaConfigOptionsToString(). Removed

scholzj · 2022-12-08T14:22:15Z

java/kafka/consumer/src/main/java/KafkaConsumerConfig.java

+    private static final Map<String, String> USER_CONFIGS = System.getenv()
+            .entrySet()
+            .stream()
+            .filter(map -> map.getKey().startsWith(KAFKA_PREFIX))
+            .collect(Collectors.toMap(map -> convertEnvVarToPropertyKey(map.getKey()), map -> map.getValue()));


Should you use Properties directly here?

I also wonder if this should be static. In general, for things such as testing etc., it would be better to handle this in the constructor or in the fromEnv method. You might nto test it today, but maybe you can add some tests for this in the future.

Discuss today with @kyguy

scholzj · 2022-12-08T14:23:59Z

java/kafka/producer/src/main/java/KafkaProducerConfig.java

+                "bootstrapServers='" + bootstrapServers + '\'' +
+                ", topic='" + topic + '\'' +
+                ", delay=" + delay +
+                ", messageCount=" + messageCount +
+                ", message='" + message + '\'' +
+                ", acks='" + acks + '\'' +
+                ", headers='" + headers + '\'' +
+                ", sslTruststoreCertificates='" + sslTruststoreCertificates + '\'' +
+                ", sslKeystoreKey='" + sslKeystoreKey + '\'' +
+                ", sslKeystoreCertificateChain='" + sslKeystoreCertificateChain + '\'' +
+                ", oauthClientId='" + oauthClientId + '\'' +
+                ", oauthClientSecret='" + oauthClientSecret + '\'' +
+                ", oauthAccessToken='" + oauthAccessToken + '\'' +
+                ", oauthRefreshToken='" + oauthRefreshToken + '\'' +
+                ", oauthTokenEndpointUri='" + oauthTokenEndpointUri + '\'' +
+                ", additionalConfig='" + additionalConfig + '\'' +
+                ", tracingSystem='" + tracingSystem + '\'' +


Maybe this should not be part of this PR? Ideally, you should try to keep the PRs focused on a single thing.

Will remove

scholzj · 2022-12-08T14:24:14Z

java/kafka/streams/src/main/java/KafkaStreamsExample.java

@@ -46,8 +46,7 @@ public static void main(String[] args) {
                KafkaClientSupplier supplier = new TracingKafkaClientSupplier();
                streams = new KafkaStreams(builder.build(), props, supplier);
            } else {
-                log.error("Error: TRACING_SYSTEM {} is not recognized or supported!", config.getTracingSystem());
-                streams = new KafkaStreams(builder.build(), props);
+                throw new RuntimeException("Error: TRACING_SYSTEM " + tracingSystem + " is not recognized or supported!");


Same as above ... Maybe this should not be part of this PR? Ideally, you should try to keep the PRs focused on a single thing.

Will remove

Revert to original

scholzj · 2022-12-08T14:25:48Z

java/kafka/consumer/src/main/java/KafkaConsumerConfig.java

    private final String topic;
-    private final String groupId;
-    private final String autoOffsetReset = "earliest";
    private final String enableAutoCommit = "false";


This seems to be a weird left-over. You do not set it, but you keep it in the code. If it is needed, maybe you should check if it is enabled or not through the Kafka properties and set it from there? (https://kafka.apache.org/documentation/#consumerconfigs_enable.auto.commit)

autoOffsetReset removed. Discuss enableAutoCommit with @kyguy

I think it is a bit more complicated then just removing it. The auto-commit is part of Kafka. But what behaviour do you expect when it is disabled? How does the client do the commits etc.? That is something you need to consider.

scholzj · 2022-12-08T14:27:44Z

README.md

+* `KAFKA_BOOTSTRAP_SERVERS` - comma-separated host and port pairs that is a list of Kafka broker addresses. The form of pair is `host:port`, e.g. `my-cluster-kafka-bootstrap:9092` 
+* `STRIMZI_TOPIC` - name of topic which consumer subscribes  
+* `KAFKA_GROUP_ID` - specifies the consumer group id for the consumer
+* `STRIMZI_MESSAGE_COUNT` - the number of messages the consumer should receive
+* `STRIMZI_LOG_LEVEL` - logging level  


You should expans this a bit. List the Strimzi specific options and elaborate that any Kafka otpions can be used as Env var by:

Preficing it with KAFKA_ and making it uppercase and using _ instead of ..

Added:
Below are listed and described environmental variables that are NOT Kafka-specific but must be used. These environmental variables should be prefixed
with uppercase STRIMZI_ using _ instead of . (eg. STRIMZI_TOPIC). Any Kafka configuration option can be used as an environmental variable and should be prefixed with uppercase KAFKA_ using _ instead of . (eg. KAFKA_BOOTSTRAP_SERVERS).

kyguy · 2022-12-13T14:59:34Z

java/kafka/consumer/src/main/java/KafkaConsumerExample.java

@@ -41,7 +43,7 @@ public static void main(String[] args) {
            }
        }

-        boolean commit = !Boolean.parseBoolean(config.getEnableAutoCommit());
+       boolean commit = !Boolean.parseBoolean (props.getProperty(ENABLE_AUTO_COMMIT_CONFIG));


Suggested change

boolean commit = !Boolean.parseBoolean (props.getProperty(ENABLE_AUTO_COMMIT_CONFIG));

boolean commit = !Boolean.parseBoolean(props.getProperty(ENABLE_AUTO_COMMIT_CONFIG));

kyguy · 2022-12-13T15:04:08Z

java/kafka/deployment-ssl.yaml

            value: java-kafka-consumer
-          - name: LOG_LEVEL
+          - name: STRIMZI_ LOG_LEVEL


Suggested change

- name: STRIMZI_ LOG_LEVEL

- name: STRIMZI_LOG_LEVEL

kyguy · 2022-12-13T15:05:12Z

java/kafka/deployment-tracing-jaeger.yaml

@@ -3,7 +3,7 @@ kind: KafkaTopic
 metadata:
  name: my-topic
  labels:
-      strimzi.io/cluster: my-cluster
+    strimzi.io/cluster: my-cluster


extra two spaces removed

kyguy · 2022-12-13T15:08:39Z

java/kafka/deployment-tracing-jaeger.yaml

+          - name: JAEGER_SAMPLER_PARAM
+            value: "1"
+          - name: TRACING_SYSTEM
+            value: jaeger


add two space indentation back

kyguy · 2022-12-13T15:10:30Z

java/kafka/streams/src/main/java/KafkaStreamsExample.java

@@ -47,8 +47,7 @@ public static void main(String[] args) {
                streams = new KafkaStreams(builder.build(), props, supplier);
            } else {
                log.error("Error: TRACING_SYSTEM {} is not recognized or supported!", config.getTracingSystem());
-                streams = new KafkaStreams(builder.build(), props);
-            }


Rogue formatting change, can be undone

kyguy

Just some formatting nits

scholzj

Left few nits mostl around formatting etc. But looks good overall.

scholzj · 2022-12-20T14:51:52Z

README.md

@@ -57,7 +57,9 @@ To run the OAuth example, you will need to have your Kafka cluster configured wi
 ## Configuration

 Although this Hello World is simple example it is fully configurable.
-Below are listed and described environmental variables.
+Below are listed and described environmental variables that are NOT Kafka-specific but must be used. 
+These environmental variables should be prefixed with `STRIMZI_` using `_` instead of `.` (eg. STRIMZI_TOPIC). 


These are just random environment variables. The _ versus . makes no sense here. It makes sense only for the KAFKA_ env vars which map to actual Kafka configuration properties.

Removed and refactored to fit the uniformity of the rest of the README.

scholzj · 2022-12-20T14:52:07Z

README.md

-Below are listed and described environmental variables.
+Below are listed and described environmental variables that are NOT Kafka-specific but must be used. 
+These environmental variables should be prefixed with `STRIMZI_` using `_` instead of `.` (eg. STRIMZI_TOPIC). 
+Any Kafka configuration option can be given as an environmental variable and should be prefixed with `KAFKA_` using `_` instead of `.` (eg. bootstrap.servers -> KAFKA_BOOTSTRAP_SERVERS).


Suggested change

Any Kafka configuration option can be given as an environmental variable and should be prefixed with `KAFKA_` using `_` instead of `.` (eg. bootstrap.servers -> KAFKA_BOOTSTRAP_SERVERS).

Any Kafka configuration option can be given as an environmental variable and should be prefixed with `KAFKA_` using `_` instead of `.` (eg. `bootstrap.servers` -> `KAFKA_BOOTSTRAP_SERVERS`).

Added back in

scholzj · 2022-12-20T14:53:16Z

README.md

+Below are listed and described environmental variables that are NOT Kafka-specific but must be used. 
+These environmental variables should be prefixed with `STRIMZI_` using `_` instead of `.` (eg. STRIMZI_TOPIC). 
+Any Kafka configuration option can be given as an environmental variable and should be prefixed with `KAFKA_` using `_` instead of `.` (eg. bootstrap.servers -> KAFKA_BOOTSTRAP_SERVERS).


So, does this currently apply to consumers only? It is fine to do it step by step. But you should make it clear here in the README. Right now it sounds like this applies for producers and Streams as well.

Refactored to specify Consumer only (until later updates)

scholzj · 2022-12-20T14:55:51Z

java/kafka/consumer/src/main/java/KafkaConsumerConfig.java

-    public String getBootstrapServers() {
-        return bootstrapServers;
+    public static String convertEnvVarToPropertyKey(String envVar) {
+        System.out.println("ENV_VAR is " + envVar);


This should use logger if you want to log this (doesn't seem needed to me, but why not) or be removed if it is some leftover from debugging.

Removed altogether

OwenCorrigan76 · 2022-12-21T09:48:54Z

@scholzj @kyguy Thanks for the reviews / feedback. I've made the latest changes as commented by @scholzj. I will now start to refactor the ProducerConfigs. As I am doing so, I think it prudent to refactor the name of the PR to Refactor Client-Example Configs.

ppatierno · 2022-12-21T12:05:29Z

README.md

@@ -57,7 +57,10 @@ To run the OAuth example, you will need to have your Kafka cluster configured wi
 ## Configuration

 Although this Hello World is simple example it is fully configurable.
-Below are listed and described environmental variables.
+Below are listed and described environmental variables. 


I think we use "environment variable" everywhere in our docs. Can we change this way everywhere you are using "environmental"?

ppatierno · 2022-12-21T12:08:18Z

java/kafka/consumer/src/main/java/KafkaConsumerConfig.java

+                .entrySet()
+                .stream()
+                .filter(map -> map.getKey().startsWith(KAFKA_PREFIX))
+                .collect(Collectors.toMap(map -> convertEnvVarToPropertyKey(map.getKey()), map -> map.getValue())));


is map appropriate as variable here? You are getting a Map.Entry instance right? I would be more for naming it entry or mapEntry but not just map which makes reader thinking about the whole map itself.

Refactored to use mapEntry

ppatierno · 2022-12-21T12:09:03Z

java/kafka/consumer/src/main/java/KafkaConsumerExample.java

@@ -41,7 +43,7 @@ public static void main(String[] args) {
            }
        }

-        boolean commit = !Boolean.parseBoolean(config.getEnableAutoCommit());
+       boolean commit = !Boolean.parseBoolean(props.getProperty(ENABLE_AUTO_COMMIT_CONFIG));


wrong indentation, missing space?

scholzj · 2022-12-21T15:01:38Z

@scholzj @kyguy Thanks for the reviews / feedback. I've made the latest changes as commented by @scholzj. I will now start to refactor the ProducerConfigs. As I am doing so, I think it prudent to refactor the name of the PR to Refactor Client-Example Configs.

Too late I guess. But we should have first approved and merged the consumer config you had. This just makes it harder to review and fix things.

kyguy · 2022-12-21T15:28:12Z

java/kafka/producer/src/main/java/KafkaProducerConfig.java

-        return props;
+        String topic = System.getenv("STRIMZI_TOPIC");
+        Long messageCount = System.getenv("STRIMZI_MESSAGE_COUNT") == null ? DEFAULT_MESSAGES_COUNT : Long.parseLong(System.getenv("STRIMZI_MESSAGE_COUNT"));
+        int delay = System.getenv("STRIMZI_DELAY") == null ? DEFAULT_DELAY : Integer.parseInt(System.getenv("STRIMZI_DELAY"));


Suggested change

int delay = System.getenv("STRIMZI_DELAY") == null ? DEFAULT_DELAY : Integer.parseInt(System.getenv("STRIMZI_DELAY"));

int delay = Integer.parseInt(System.getenv("STRIMZI_DELAY_MS"));

kyguy · 2022-12-21T15:33:08Z

java/kafka/producer/src/main/java/KafkaProducerConfig.java


 public class KafkaProducerConfig {
    private static final Logger log = LogManager.getLogger(KafkaProducerConfig.class);

    private static final long DEFAULT_MESSAGES_COUNT = 10;
    private static final String DEFAULT_MESSAGE = "Hello world";
-    private final String bootstrapServers;
+    private static final int DEFAULT_DELAY = 1;


Can be removed

kyguy · 2022-12-21T15:36:33Z

java/kafka/producer/src/main/java/KafkaProducerExample.java

@@ -19,6 +19,7 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicLong;

+


kyguy · 2022-12-21T15:37:35Z

java/kafka/producer/src/main/java/KafkaProducerExample.java

-        boolean transactionalProducer = config.getAdditionalConfig().contains("transactional.id");
+      //  boolean transactionalProducer = !Boolean.parseBoolean(props.getProperty(TRANSACTIONAL_ID_CONFIG));
+        boolean transactionalProducer = config.getProperties().contains("transactional.id");
+      //  boolean transactionalProducer = config.getAdditionalConfig().contains("transactional.id");


Don't forget to clean up

Reverted to before refactoring ConsumerConfig

kyguy · 2022-12-21T15:39:59Z

java/kafka/streams/src/main/java/KafkaStreamsExample.java

@@ -46,8 +46,7 @@ public static void main(String[] args) {
                KafkaClientSupplier supplier = new TracingKafkaClientSupplier();
                streams = new KafkaStreams(builder.build(), props, supplier);
            } else {
-                log.error("Error: TRACING_SYSTEM {} is not recognized or supported!", config.getTracingSystem());
-                streams = new KafkaStreams(builder.build(), props);
+                throw new RuntimeException("Error: TRACING_SYSTEM " + tracingSystem + " is not recognized or supported!");


Revert to original

kyguy · 2022-12-21T15:40:17Z

java/kafka/streams/src/main/java/KafkaStreamsExample.java

@@ -46,13 +46,12 @@ public static void main(String[] args) {
                KafkaClientSupplier supplier = new TracingKafkaClientSupplier();
                streams = new KafkaStreams(builder.build(), props, supplier);
            } else {
-                log.error("Error: TRACING_SYSTEM {} is not recognized or supported!", config.getTracingSystem());
-                streams = new KafkaStreams(builder.build(), props);


Add back in, revert to original

Signed-off-by: Owen <owencorrigan76@gmail.com>

kyguy · 2022-12-21T16:31:26Z

java/kafka/streams/src/main/java/KafkaStreamsExample.java

@@ -54,4 +54,4 @@ public static void main(String[] args) {

        streams.start();
    }
-}


ppatierno · 2022-12-21T16:33:57Z

@OwenCorrigan76 can we move out of the draft state?

ppatierno · 2022-12-21T16:35:30Z

Agree with @scholzj ... but it seems not to be late yet. There are no changes on the producer side yet :-)

Signed-off-by: Owen <owencorrigan76@gmail.com>

OwenCorrigan76 · 2022-12-22T10:03:19Z

@scholzj @ppatierno I had done some work on ProducerConfigs but I reverted everything back to before I changed ProducerConfigs. I guess I'll concentrate on Consumer first. Get it merged and then open a new PR for Producer. When that is merged, I'll open a separate PR for Streams. This means that any of Kyle's comments regarding ProducerConfigs above can be ignored. I know it's messy but i originally thought you guys would prefer Producer / Consumer and Streams done in the same PR (my mistake).

scholzj · 2022-12-21T20:37:50Z

java/kafka/consumer/src/main/java/KafkaConsumerConfig.java

-
-    public String getBootstrapServers() {
-        return bootstrapServers;
+    public static String convertEnvVarToPropertyKey(String envVar) {


Can be private I guess?

scholzj · 2022-12-21T20:40:39Z

README.md

@@ -57,7 +57,10 @@ To run the OAuth example, you will need to have your Kafka cluster configured wi
 ## Configuration

 Although this Hello World is simple example it is fully configurable.
-Below are listed and described environmental variables.
+Below are listed and described environment variables. 
+Regarding Consumer Only (until later updates): below are listed and described environmental variables that are NOT Kafka-specific but must be used. 


Only with lowercase o I guess? Also, is must be used the right way?

scholzj · 2022-12-21T20:47:24Z

README.md

@@ -57,7 +57,10 @@ To run the OAuth example, you will need to have your Kafka cluster configured wi
 ## Configuration


Maybe you can do something like this (just the Configuration section):

## Configuration Below are listed and described the available environment variables that can be used for configuration. ### Producer * `BOOTSTRAP_SERVERS` - comma-separated host and port pairs that is a list of Kafka broker addresses. The form of pair is `host:port`, e.g. `my-cluster-kafka-bootstrap:9092` * `TOPIC` - the topic the producer will send to * `DELAY_MS` - the delay, in ms, between messages * `MESSAGE_COUNT` - the number of messages the producer should send * `CA_CRT` - the certificate of the CA which signed the brokers' TLS certificates, for adding to the client's trust store * `USER_CRT` - the user's certificate * `USER_KEY` - the user's private key * `LOG_LEVEL` - logging level * `PRODUCER_ACKS` - acknowledgement level * `HEADERS` - custom headers list separated by commas of `key1=value1, key2=value2` * `BLOCKING_PRODUCER` - if it's set, the producer will block another message until ack will be received * `MESSAGES_PER_TRANSACTION` - how many messages will be part of one transaction. Transaction config could be set via `ADDITIONAL_CONFIG` variable. Default is 10. * `ADDITIONAL_CONFIG` - additional configuration for a producer application. Notice, that you can also override any previously set variable by setting this. The form is `key=value` records separated by new line character * `TRACING_SYSTEM` - if it's set to `jaeger` or `opentelemetry`, this will enable tracing. ### Consumer * `STRIMZI_TOPIC` - name of topic to which consumer subscribes * `STRIMZI_MESSAGE_COUNT` - the number of messages the consumer should receive * `STRIMZI_LOG_LEVEL` - logging level * `STRIMZI_TRACING_SYSTEM` - if it's set to `jaeger` or `opentelemetry`, this will enable tracing. Additionally, any Kafka Consumer API configuration option can be passed as an environmental variable. It should be prefixed with `KAFKA_` and use `_` instead of `.`. For example environment variable `KAFKA_BOOTSTRAP_SERVERS` will be used as the `bootstrap.servers` configuration option in the Kafka Consumer API. ### Streams * `BOOTSTRAP_SERVERS` - comma-separated host and port pairs that is a list of Kafka broker addresses. The form of pair is `host:port`, e.g. `my-cluster-kafka-bootstrap:9092` * `APPLICATION_ID` - The Kafka Streams application ID * `SOURCE_TOPIC` - name of topic which will be used as the source of messages * `TARGET_TOPIC` - name of topic where the transformed images are sent * `COMMIT_INTERVAL_MS` - the interval for the Kafka Streams consumer part committing the offsets * `CA_CRT` - the certificate of the CA which signed the brokers' TLS certificates, for adding to the client's trust store * `USER_CRT` - the user's certificate * `USER_KEY` - the user's private key * `LOG_LEVEL` - logging level * `ADDITIONAL_CONFIG` - additional configuration for a streams application. Notice, that you can also override any previously set variable by setting this. The form is `key=value` records separated by new line character. * `TRACING_SYSTEM` - if it's set to `jaeger` or `opentelemetry`, this will enable tracing. ### Tracing The examples support tracing using the [OpenTracing Apache Kafka Instrumentation](https://github.com/opentracing-contrib/java-kafka-client), [OpenTelemetry Java Instrumentation](https://github.com/open-telemetry/opentelemetry-java-instrumentation) and the [Jaeger project](https://www.jaegertracing.io/). To enable tracing, configure the Jaeger Tracer using [environment variables](https://github.com/jaegertracing/jaeger-client-java/tree/master/jaeger-core#configuration-via-environment). To run Jaeger Tracing, you can also use the provided example in [`deployment-tracing-jaeger.yaml`](./java/kafka/deployment-tracing-jaeger.yaml). To run OpenTelemetry Tracing, you can also use the provided example in [`deployment-tracing-opentelemetry.yaml`](./java/kafka/deployment-tracing-opentelemetry.yaml). Jaeger / OpenTracing tracing is supported only in consumers / producers because OpenTracing support for Kafka Streams API is not compatible with the latest Kafka versions.

Makes sense. Refactored to use this suggestion.

Signed-off-by: Owen <owencorrigan76@gmail.com>

scholzj

Thanks for the PR!

This reverts commit 0362308.

* for refactoring Signed-off-by: Owen <owencorrigan76@gmail.com> * refactored ConsumerConfigs Signed-off-by: Owen <owencorrigan76@gmail.com> * refactored ConsumerConfigs Signed-off-by: Owen <owencorrigan76@gmail.com> * removed Ssl and Oauth configs Signed-off-by: Owen <owencorrigan76@gmail.com> * Addressing Comments Signed-off-by: Owen <owencorrigan76@gmail.com> * Indentation in jaeger.yaml Signed-off-by: Owen <owencorrigan76@gmail.com> * Addressing Kyle's Comments Signed-off-by: Owen <owencorrigan76@gmail.com> * Addressing Jakub's Comments and changing Repo name Signed-off-by: Owen <owencorrigan76@gmail.com> * Addressing Paolo's Comments Signed-off-by: Owen <owencorrigan76@gmail.com> * Addressing Jakub and Paolo's Comments Signed-off-by: Owen <owencorrigan76@gmail.com> * Further Addressing Jakub's Comments Signed-off-by: Owen <owencorrigan76@gmail.com> Signed-off-by: Owen <owencorrigan76@gmail.com>

…i#94) This reverts commit 0362308.

OwenCorrigan76 assigned kyguy Dec 6, 2022

scholzj reviewed Dec 6, 2022

View reviewed changes

kyguy reviewed Dec 6, 2022

View reviewed changes

OwenCorrigan76 self-assigned this Dec 7, 2022

scholzj reviewed Dec 8, 2022

View reviewed changes

OwenCorrigan76 requested review from scholzj and kyguy December 13, 2022 10:38

kyguy reviewed Dec 13, 2022

View reviewed changes

OwenCorrigan76 requested a review from ppatierno December 15, 2022 10:02

scholzj reviewed Dec 20, 2022

View reviewed changes

OwenCorrigan76 changed the title ~~Refactor consumer configs~~ Refactor Client-Example configs Dec 21, 2022

OwenCorrigan76 removed the request for review from ppatierno December 21, 2022 09:51

ppatierno reviewed Dec 21, 2022

View reviewed changes

kyguy reviewed Dec 21, 2022

View reviewed changes

OwenCorrigan76 force-pushed the RefactorConsumerConfigs branch from be1363d to 8993420 Compare December 21, 2022 15:55

OwenCorrigan76 changed the title ~~Refactor Client-Example configs~~ Refactor Client_Example ConsumerConfigs Dec 21, 2022

OwenCorrigan76 changed the title ~~Refactor Client_Example ConsumerConfigs~~ Refactor Client-Example ConsumerConfigs Dec 21, 2022

OwenCorrigan76 added 3 commits December 21, 2022 16:06

for refactoring

Verified

This commit was created on GitHub.com and signed with GitHub’s verified signature.

GPG key ID: B5690EEEBB952194

Verified
Learn about vigilant mode

a50133d

Signed-off-by: Owen <owencorrigan76@gmail.com>

refactored ConsumerConfigs

0b6f700

Signed-off-by: Owen <owencorrigan76@gmail.com>

refactored ConsumerConfigs

2468839

Signed-off-by: Owen <owencorrigan76@gmail.com>

OwenCorrigan76 added 6 commits December 21, 2022 16:06

removed Ssl and Oauth configs

daa7a94

Signed-off-by: Owen <owencorrigan76@gmail.com>

Addressing Comments

2fe3e8e

Signed-off-by: Owen <owencorrigan76@gmail.com>

Indentation in jaeger.yaml

23ce430

Signed-off-by: Owen <owencorrigan76@gmail.com>

Addressing Kyle's Comments

e89770a

Signed-off-by: Owen <owencorrigan76@gmail.com>

Addressing Jakub's Comments and changing Repo name

d303343

Signed-off-by: Owen <owencorrigan76@gmail.com>

Addressing Paolo's Comments

13836f0

Signed-off-by: Owen <owencorrigan76@gmail.com>

OwenCorrigan76 force-pushed the RefactorConsumerConfigs branch from 8993420 to 13836f0 Compare December 21, 2022 16:18

kyguy reviewed Dec 21, 2022

View reviewed changes

kyguy approved these changes Dec 21, 2022

View reviewed changes

OwenCorrigan76 marked this pull request as ready for review December 22, 2022 09:47

Addressing Jakub and Paolo's Comments

17341e4

Signed-off-by: Owen <owencorrigan76@gmail.com>

OwenCorrigan76 requested review from scholzj and ppatierno December 22, 2022 10:03

ppatierno approved these changes Dec 22, 2022

View reviewed changes

scholzj reviewed Dec 22, 2022

View reviewed changes

Further Addressing Jakub's Comments

e7e7bdd

Signed-off-by: Owen <owencorrigan76@gmail.com>

OwenCorrigan76 requested a review from scholzj December 22, 2022 14:40

scholzj approved these changes Dec 22, 2022

View reviewed changes

scholzj merged commit 0362308 into strimzi:main Dec 22, 2022

scholzj added a commit that referenced this pull request Dec 23, 2022

Revert "Refactor Client-Example ConsumerConfigs (#91)"

f2ee89f

This reverts commit 0362308.

scholzj mentioned this pull request Dec 23, 2022

Revert "Refactor Client-Example ConsumerConfigs" #94

Merged

scholzj added a commit that referenced this pull request Dec 23, 2022

Revert "Refactor Client-Example ConsumerConfigs (#91)" (#94)

8613242

This reverts commit 0362308.

OwenCorrigan76 pushed a commit to OwenCorrigan76/client-examples that referenced this pull request Jan 30, 2024

Revert "Refactor Client-Example ConsumerConfigs (strimzi#91)" (strimz…

17e791f

…i#94) This reverts commit 0362308.

	boolean commit = !Boolean.parseBoolean (props.getProperty(ENABLE_AUTO_COMMIT_CONFIG));
	boolean commit = !Boolean.parseBoolean(props.getProperty(ENABLE_AUTO_COMMIT_CONFIG));

	Any Kafka configuration option can be given as an environmental variable and should be prefixed with `KAFKA_` using `_` instead of `.` (eg. bootstrap.servers -> KAFKA_BOOTSTRAP_SERVERS).
	Any Kafka configuration option can be given as an environmental variable and should be prefixed with `KAFKA_` using `_` instead of `.` (eg. `bootstrap.servers` -> `KAFKA_BOOTSTRAP_SERVERS`).

	int delay = System.getenv("STRIMZI_DELAY") == null ? DEFAULT_DELAY : Integer.parseInt(System.getenv("STRIMZI_DELAY"));
	int delay = Integer.parseInt(System.getenv("STRIMZI_DELAY_MS"));

		@@ -19,6 +19,7 @@
		import java.util.concurrent.Future;
		import java.util.concurrent.atomic.AtomicLong;

		@@ -57,7 +57,10 @@ To run the OAuth example, you will need to have your Kafka cluster configured wi
		## Configuration

Refactor Client-Example ConsumerConfigs #91

Refactor Client-Example ConsumerConfigs #91

Conversation

OwenCorrigan76 commented Dec 6, 2022 • edited Loading

scholzj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

OwenCorrigan76 commented Dec 7, 2022

scholzj commented Dec 7, 2022

OwenCorrigan76 commented Dec 8, 2022

scholzj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

OwenCorrigan76 Dec 12, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

OwenCorrigan76 Dec 12, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kyguy Dec 13, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kyguy left a comment

Choose a reason for hiding this comment

scholzj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

OwenCorrigan76 Dec 21, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

OwenCorrigan76 commented Dec 21, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

scholzj commented Dec 21, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ppatierno commented Dec 21, 2022

ppatierno commented Dec 21, 2022

OwenCorrigan76 commented Dec 22, 2022

OwenCorrigan76 commented Dec 6, 2022 •

edited

Loading

OwenCorrigan76 Dec 12, 2022 •

edited

Loading

OwenCorrigan76 Dec 12, 2022 •

edited

Loading

kyguy Dec 13, 2022 •

edited

Loading

OwenCorrigan76 Dec 21, 2022 •

edited

Loading

OwenCorrigan76 Dec 22, 2022 •

edited

Loading

OwenCorrigan76 Dec 22, 2022 •

edited

Loading