Allow bootstrap.servers to be provided for Kafka ingestion.

[JulianJaffePinterest]

Addresses #8685, allowing for bootstrap.servers to be provided by a PasswordProvider instead of hard-coded.

Description

Modified KafkaRecordSupplier.addConsumerPropertiesFromConfig to check for a password provider when deserializing bootstrap.servers as it currently does when deserializing trust and key store passwords. This does not implement the further suggestion on #8685 to allow any field to be provided by a PasswordProvider because #6666 is still an open issue.

---

This PR has:

• been self-reviewed.
• added documentation for new or modified features or behaviors.
• added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added or updated version, license, or notice information in licenses.yaml
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths.
• added integration tests.
• been tested in a test Druid cluster.

---

Key changed/added classes in this PR

• KafkaRecordSupplier

[gianm]

Sorry for the delay in getting back to you. I think we need to add a bit more to this patch, as mentioned in the review comments.

[gianm]

This part is going to throw a wrench into backwards compatibility (we'll be writing ioConfigs that prior versions can't read). It might negatively affect rolling upgrades.

I think the easiest way to solve this is to adjust DefaultPasswordProvider so it serializes as a simple string. (It already is able to deserialize itself from a simple string, so no problem on that end.) The way to do that is add a @JsonValue annotation to the getPassword() method.

[JulianJaffePinterest]

Ah that's a good point (in order to get the DefaultPasswordProvider to serialize back to a simple string I think we also need to add @JsonTypeInfo(use = JsonTypeInfo.Id.NONE) so it won't serialize the type info). The tricky piece here is not breaking the PasswordProviderRedactionMixIn, as it seems that @JsonValue takes precedence over @JsonIgnore, even when I set @JsonValue(false) in the redaction mixin. I'll whip up a custom serializer to handle this.

[gianm]

Excellent point about the redaction mixin. Looking forward to what you come up with.

[gianm]

A bit of an abuse of the "password provider" name, but I can see why it's useful. Perhaps, in the future, we'd like to change the name to something more generic. I wouldn't do that in this patch though. (Each patch should just do one thing, ideally.)

[JulianJaffePinterest]

Yeah, part of #9351 is moving to CredentialProvider from PasswordProvider, which seems like a better term for this use case as well. I suppose eventually it might make sense to just go with Provider or the like if we want to allow any field to be provided at runtime.

[stale]

This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 4 weeks if no further activity occurs. If you think that's incorrect or this pull request should instead be reviewed, please simply write any comment. Even if closed, you can still revive the PR at any time or discuss it on the dev@druid.apache.org list. Thank you for your contributions.

[himanshug]

I have an alternate implementation proposal of achieving this.

Re #9351 : We can still introduce the new interface and use that in this use case. Migrating all other places from PasswordProvider to new interface could be taken up in #9351 . No reason for continuing to use PasswordProvider when there is general agreement on the newer interface.
However, considering this use case, I would name new interface DynamicConfigProvider instead of CredentialProvider .

Then KafkaRecordSupplier code change would end up looking something like ...

  public static void addConsumerPropertiesFromConfig(
      Properties properties,
      ObjectMapper configMapper,
      Map<String, Object> consumerProperties
  )
  {
    Object dynamicConfigProviderJson = consumerProperties.get("DRUID_DYNAMIC_CONFIG_PROVIDER_KEY");
    if (dynamicConfigProviderJson != null) {
      DynamicConfigProvider dynamicConfigProvider = configMapper.convertValue(dynamicConfigProviderJson, DynamicConfigProvider.class);
      Map<String, String> dynamicConfig = dynamicConfigProvider.getConfig();

      for (Map.Entry<String, Object> entry : consumerProperties.entrySet()) {
        String propertyKey = entry.getKey();

        if ("DRUID_DYNAMIC_CONFIG_PROVIDER_KEY".equals(propertyKey)) continue; //skip the dynamicConfig entry

        String properyValue = String.valueOf(entry.getValue());
        String valueFromDynamicConfig = dynamicConfig.get(properyValue);

        if (valueFromDynamicConfig == null) {
          properties.setProperty(propertyKey, properyValue);
        } else {
          properties.setProperty(propertyKey, valueFromDynamicConfig);
        }
      }
    } else {
      // this is required for backward compabitibility.
      for (Map.Entry<String, Object> entry : consumerProperties.entrySet()) {
        String propertyKey = entry.getKey();
        if (propertyKey.equals(KafkaSupervisorIOConfig.TRUST_STORE_PASSWORD_KEY)
            || propertyKey.equals(KafkaSupervisorIOConfig.KEY_STORE_PASSWORD_KEY)
            || propertyKey.equals(KafkaSupervisorIOConfig.KEY_PASSWORD_KEY)) {
          PasswordProvider configPasswordProvider = configMapper.convertValue(
              entry.getValue(),
              PasswordProvider.class
          );
          properties.setProperty(propertyKey, configPasswordProvider.getPassword());
        } else {
          properties.setProperty(propertyKey, String.valueOf(entry.getValue()));
        }
      }
    }
  }

which maintains backward compatibility as well which is a concern raised in another comment.

@gianm @JulianJaffePinterest does that sound reasonable ?

[stale]

This issue is no longer marked as stale.

[himanshug]

@JulianJaffePinterest are you still planning to work on this? I need bootstrap server dynamic configurability as well. So, I can probably finish it off if you want.

[himanshug]

closing this in favor of #10309 which makes all kafka consumer props including bootstrap.servers to be provided in a dynamic way via extensions.