HPCC-33304 Kafka plugin: Support custom client configurations in Helm chart #19472

dcamper · 2025-02-03T17:49:40Z

Type of change:

This change is a bug fix (non-breaking change which fixes an issue).
This change is a new feature (non-breaking change which adds functionality).
This change improves the code (refactor or other change that does not change the functionality)
This change fixes warnings (the fix does not alter the functionality or the generated code)
This change is a breaking change (fix or feature that will cause existing behavior to change).
This change alters the query API (existing queries will have to be recompiled)

Checklist:

Smoketest:

Send notifications about my Pull Request position in Smoketest queue.
Test my draft Pull Request.

Testing:

Manual testing to ensure that helm chart is configured correctly and the plugin code can read the Kafka configuration options there.

github-actions · 2025-02-03T17:49:59Z

Jira Issue: https://hpccsystems.atlassian.net//browse/HPCC-33304

Jirabot Action Result:
Workflow Transition To: Merge Pending
Updated PR

jakesmith

@dcamper - looks like there are some compilation issues.

jakesmith · 2025-02-06T15:16:14Z

helm/hpcc/values.schema.json

@@ -1728,6 +1733,11 @@
          "default": false,
          "description": "Require SOAPCALL and HTTPCALL URLs are secrets or mapped to secrets"
        },
+        "plugins": {


this (and others in other engines) should reuse the "definitions.plugins" you've added at line 434
e.g.:

"plugins": { "$ref": "#/definitions/plugins" },

Done, though I admit that I find this syntax confusing.

jakesmith · 2025-02-06T15:20:57Z

plugins/kafka/README.md

@@ -345,7 +440,8 @@ long as needed.

 ### Saved Topic Offsets

-By default, consumers save to a file the offset of the last-read message from a
+By default, in a bare-metal environment, consumers save to a file the offset of the
+last-read message from a


does the newline/wrap need adjusting?

Fixed here as well as the other one. That's what I get for staring at the Markdown rendered version.

jakesmith · 2025-02-06T15:21:12Z

plugins/kafka/README.md

@@ -356,8 +452,11 @@ exact path may be different if you have named an engine differently in your HPCC
 configuration).  The format of the saved offset filename is
 `<TopicName>-<PartitionNum>-<ConsumerGroup>.offset`.

-Note that saving partition offsets is engine-specific.  One practical
+Note that saving partition offsets is engine-specific in a bare-metal environment.
+One practical


jakesmith

@dcamper - please see comments.

jakesmith · 2025-02-06T16:10:56Z

plugins/kafka/kafka.cpp

@@ -54,24 +58,80 @@ namespace KafkaPlugin
    // Static Methods (internal)
    //--------------------------------------------------------------------------

+#ifdef _CONTAINERIZED


we normally try to compile both code alternatives, by using isContainerized(), unless we can't because it's dependent on bits of the system that are [e.g.] not in containerized.
With some hope that eventually almost all the code base will follow the 'containerized' approach.

In this case.. I guess it's too late for the code to be common?
e.g. could the BM system use yaml config files, and thus this applyConfigProps be common? Prob. not.

Unfortunately, I don't think that would work. We do have active users on the bare metal side of things, so that configuration layout and usage of local files needs to stay in place.

I thought so, but wanted to check.
Avoiding the #ifdef _CONTAINERIZED I think still worthwhile though. It prevents hitting issues like the compile problem as both code paths are being compiled either way (hopefully there isn't too much code bloat, and BM code will slowly fade away).

jakesmith · 2025-02-06T16:11:49Z

plugins/kafka/kafka.cpp

+
+                if (name && *name)
+                {
+                    if (stricmp(name, "metadata.broker.list") != 0)


trivial: clearer to combine the 2 ifs statements and use strsame

Agreed, but the else from the first if() leads to a DBGLOG that is useful. I did change the second if() to use !strisame() though. Also changed similar test on line ~146.

DBGLOG may be enough, but at some point we aim to turn them off, so should they be OWARN or similar? (or will that be too verbose

Excellent point. I converted a few DBGLOG calls to OWARNLOG.

jakesmith · 2025-02-06T16:12:11Z

plugins/kafka/kafka.cpp

+                {
+                    if (stricmp(name, "metadata.broker.list") != 0)
+                    {
+                        if (value && *value)


trivial: !isEmptyString ?

I like it; done (in both places).

jakesmith · 2025-02-06T16:13:38Z

plugins/kafka/kafka.cpp

+        if (configName && *configName && configPtr)
+        {
+            applyConfigProps(getGlobalConfigSP()->getPropTree("plugins/kafka"), configName, configPtr);
+            applyConfigProps(getComponentConfigSP()->getPropTree("plugins/kafka"), configName, configPtr);


is there ever any expense with calling RdKafka::Conf->set ? And if so, would it be better to compile (merge) the global/component configs, and call applyConfigProps just once ?

e.g. could do:

Owned<IPropertyTree> globalCopy = createPTreeFromIPT(getGlobalConfigSP()->queryPropTree("plugins/kafka")); synchronizePTree(globalCopy, getComponentConfigSP()->queryPropTree("plugins/kafka"), false); applyConfigProps(globalCopy, configName, configPtr);

There is very little expense in Conf->set; it is updating some C struct values. I would leave this version as-is, especially since it is not on a hot path (used only during setup, not when passing messages).

jakesmith · 2025-02-06T16:14:32Z

plugins/kafka/kafka.cpp

+    {
+        if (configName && *configName && configPtr)
+        {
+            applyConfigProps(getGlobalConfigSP()->getPropTree("plugins/kafka"), configName, configPtr);


this is leaking the IPT from getPropTree, should be queryPropTree

jakesmith · 2025-02-06T16:15:00Z

plugins/kafka/kafka.cpp

@@ -54,24 +58,80 @@ namespace KafkaPlugin
    // Static Methods (internal)
    //--------------------------------------------------------------------------

+#ifdef _CONTAINERIZED
+    static void applyConfigProps(IPropertyTree * props, const char* configName, RdKafka::Conf* configPtr)


trivial: perhaps make 'props' const for clarity

Good idea; done.

jakesmith · 2025-02-06T16:19:36Z

plugins/kafka/kafka.cpp

@@ -495,6 +560,7 @@ namespace KafkaPlugin
        consumerPtr = NULL;
        topicPtr = NULL;

+#ifndef _CONTAINERIZED
        char cpath[_MAX_DIR];

        GetCurrentDirectory(_MAX_DIR, cpath);


I think you'll get a compiler warning because not checking the return value for this (getpwd).
(we validate return elsewhere)

I added a check with an exception if the cwd couldn't be found.

jakesmith · 2025-02-06T16:20:56Z

plugins/kafka/kafka.cpp

+     * @param   configPtr           A pointer to the configuration object that
+     *                              will receive any found parameters
+     */
+    static void applyConfig(const char* configName, RdKafka::Conf* configPtr)


rather than having 2 'static void applyConfig(const char* configName, RdKafka::Conf* configPtr)' definitions, I'd vote for having the some isContainerized() code 1.

jakesmith · 2025-02-06T16:22:23Z

plugins/kafka/kafka.hpp

                /**
                 * If the offset file does not exist, create one with a
                 * default offset
                 */
-                void initFileOffsetIfNotExist() const;
+                void _initFileOffsetIfNotExist() const;


looks like this is what's causing the compile error issues (in BM).

Well that was a rookie mistake. Fixed.

jakesmith

Not for now, but meant to ask.
How long are the Consumer/Publisher instances expected to last?
Max = lifetime of a query?

If beyond the lifetime of a query, then might need to think about refreshing the configuration, when a component configuration is pushed.
We have 'installConfigUpdateHook' (and CConfigUpdateHook) for that purpose, registered and called back to perform update when the config changes.

dcamper · 2025-02-06T17:15:21Z

Not for now, but meant to ask. How long are the Consumer/Publisher instances expected to last? Max = lifetime of a query?

If beyond the lifetime of a query, then might need to think about refreshing the configuration, when a component configuration is pushed. We have 'installConfigUpdateHook' (and CConfigUpdateHook) for that purpose, registered and called back to perform update when the config changes.

Publishers can hang around until module destruction; the underlying library maintains its own cache with another thread to push messages, so it needs time to drain pending messages. Consumers last only as long as the query. I suspect that this is correct behavior and that pushed (containerized) config changes would get picked up during the next query.

jakesmith

@dcamper - looks good. Please squash.

Is 9.8 the right target (needed there?) or is 9.10 more suitable?

dcamper · 2025-02-06T18:43:26Z

@dcamper - looks good. Please squash.

Is 9.8 the right target (needed there?) or is 9.10 more suitable?

9.8 if possible, for external customers who have not yet migrated to 9.10 but are trying containerized. There is no new functionality in this change, just pulling the same configurations from a different source.

I will defer to you however and rebase if 9.10 is more appropriate.

Squashed.

… chart

jakesmith · 2025-02-07T10:14:09Z

helm/hpcc/values.schema.json

@@ -431,6 +431,11 @@
          "type": "object",
          "additionalProperties": { "type": "string" },
          "description": "Global component annotations, generated into all components"
+        },
+        "plugins": {


this definition is under "globals", it should be under "definitions".
And this should be a reference to the definition just as it is under the engine schemas.

jakesmith

@dcamper - 2 of the github action checks failing are relevant - both are related to the values schema.

jakesmith · 2025-02-07T14:14:10Z

helm/hpcc/values.schema.json

@@ -3511,6 +3523,11 @@
    "expert": {
      "description": "Settings for developers, debugging and testing",
      "type": "object"
-    }
+    },
+	"plugins": {


indentation level is off.

dcamper · 2025-02-07T16:05:52Z

Back to you, Jake. There are two failing tests but AFAICS they don't relate to these changes.

dcamper requested a review from jakesmith February 3, 2025 17:49

jakesmith reviewed Feb 6, 2025

View reviewed changes

dcamper requested a review from jakesmith February 6, 2025 16:13

jakesmith reviewed Feb 6, 2025

View reviewed changes

dcamper requested a review from jakesmith February 6, 2025 18:00

jakesmith approved these changes Feb 6, 2025

View reviewed changes

dcamper force-pushed the hpcc-33304-containerized-kafka branch from 1a72522 to f1dca9b Compare February 6, 2025 18:41

HPCC-33304 Kafka plugin: Support custom client configurations in Helm…

9e8b0dd

… chart

dcamper force-pushed the hpcc-33304-containerized-kafka branch from f1dca9b to 9e8b0dd Compare February 6, 2025 19:07

jakesmith self-requested a review February 7, 2025 10:11

jakesmith reviewed Feb 7, 2025

View reviewed changes

jakesmith requested changes Feb 7, 2025

View reviewed changes

Move plugins definition

8ff428b

jakesmith reviewed Feb 7, 2025

View reviewed changes

Don't use tabs!

d843fa0

dcamper requested a review from jakesmith February 7, 2025 16:06

HPCC-33304 Kafka plugin: Support custom client configurations in Helm chart #19472

Are you sure you want to change the base?

HPCC-33304 Kafka plugin: Support custom client configurations in Helm chart #19472

Conversation

dcamper commented Feb 3, 2025

Type of change:

Checklist:

Smoketest:

Testing:

github-actions bot commented Feb 3, 2025

jakesmith 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

jakesmith 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

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

jakesmith left a comment

Choose a reason for hiding this comment

dcamper commented Feb 6, 2025

jakesmith left a comment

Choose a reason for hiding this comment

dcamper commented Feb 6, 2025

Choose a reason for hiding this comment

jakesmith left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dcamper commented Feb 7, 2025