markap14 commented on code in PR #11399:
URL: https://github.com/apache/nifi/pull/11399#discussion_r3659183165
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java:
##########
@@ -378,41 +380,71 @@ public void inheritConfiguration(final
List<VersionedConfigurationStep> activeCo
final Bundle flowContextBundle) throws FlowUpdateException {
logger.debug("Inheriting configuration for {}", this);
- final MutableConnectorConfigurationContext configurationContext =
createConfigurationContext(activeConfig);
+
+ // Give the Connector a chance to evolve its persisted property/step
names before we build the runtime
+ // configuration contexts. Active and working configs are migrated
independently because they can diverge.
+ final Map<String, Map<String, ConnectorValueReference>>
migratedActiveProperties = migrateProperties(activeConfig);
+ final Map<String, Map<String, ConnectorValueReference>>
migratedWorkingProperties = migrateProperties(workingConfig);
+
+ final MutableConnectorConfigurationContext activeSeedContext =
createConfigurationContext(migratedActiveProperties);
final FrameworkFlowContext inheritContext =
flowContextFactory.createWorkingFlowContext(identifier,
- connectorDetails.getComponentLog(), configurationContext,
flowContextBundle);
+ connectorDetails.getComponentLog(), activeSeedContext,
flowContextBundle);
- // Apply the update for the active config
+ // Apply the active configuration. This restores activeFlowContext to
migratedActiveProperties and internally
+ // rebuilds workingFlowContext aliased to activeFlowContext's
configuration; we discard that alias below and
+ // construct an independent working context so active and working do
not share configuration state when the
+ // two lists actually diverge.
applyUpdate(inheritContext);
- // Configure the working config but do not apply
- for (final VersionedConfigurationStep step : workingConfig) {
- final StepConfiguration stepConfig = createStepConfiguration(step);
- setConfiguration(step.getName(), stepConfig, true);
+ // Tear down the working context that applyUpdate created aliased to
active, and rebuild it around an
+ // independent configuration seeded from migratedWorkingProperties.
Then fire onConfigurationStepConfigured
+ // for every step so renamed steps trigger the flow-builder callback
under their new name and any
+ // value-derived flow state (resolved asset paths, secret values,
etc.) is populated against the fresh
+ // working context.
+ destroyWorkingContext();
+ final MutableConnectorConfigurationContext workingConfigContext =
createConfigurationContext(migratedWorkingProperties);
+ workingFlowContext =
flowContextFactory.createWorkingFlowContext(identifier,
+ connectorDetails.getComponentLog(), workingConfigContext,
flowContextBundle);
+ getComponentLog().info("Working Flow Context has been rebuilt with
independent configuration");
+ for (final String stepName : migratedWorkingProperties.keySet()) {
+ notifyStepConfigured(stepName);
}
logger.debug("Successfully inherited configuration for {}", this);
}
- private StepConfiguration createStepConfiguration(final
VersionedConfigurationStep step) {
+ private Map<String, Map<String, ConnectorValueReference>>
migrateProperties(final List<VersionedConfigurationStep> flowConfiguration) {
Review Comment:
As noted above, probably worth considering an abstraction, rather than Maps
of Maps.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/migration/StandardConnectorPropertyConfiguration.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.migration;
+
+import org.apache.nifi.components.connector.ConnectorValueReference;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Framework implementation of {@link ConnectorPropertyConfiguration}. Holds
the per-step property map for a single
+ * {@link org.apache.nifi.components.connector.Connector Connector}
configuration snapshot (active or working) and
+ * tracks whether any modification has been made so callers can conditionally
persist the migrated state.
+ *
+ * <p>
+ * Instances are not thread-safe; a Connector's {@code migrateProperties}
invocation is expected to be
+ * single-threaded.
+ * </p>
+ */
+public class StandardConnectorPropertyConfiguration implements
ConnectorPropertyConfiguration {
+
+ private static final Logger logger =
LoggerFactory.getLogger(StandardConnectorPropertyConfiguration.class);
+
+ private final Map<String, Map<String, ConnectorValueReference>>
stepProperties;
+ private final Set<String> modifiedStepNames = new HashSet<>();
+ private final String componentDescription;
+ private boolean modified = false;
+
+ public StandardConnectorPropertyConfiguration(final Map<String,
Map<String, ConnectorValueReference>> initialProperties, final String
componentDescription) {
Review Comment:
Might be worth creating an abstraction here instead of using `Map<String,
Map<String, ConnectorValueReference>>` - generics whose types are generics,
especially Map within Map get a little hairy to deal with.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java:
##########
@@ -378,41 +380,71 @@ public void inheritConfiguration(final
List<VersionedConfigurationStep> activeCo
final Bundle flowContextBundle) throws FlowUpdateException {
logger.debug("Inheriting configuration for {}", this);
- final MutableConnectorConfigurationContext configurationContext =
createConfigurationContext(activeConfig);
+
+ // Give the Connector a chance to evolve its persisted property/step
names before we build the runtime
+ // configuration contexts. Active and working configs are migrated
independently because they can diverge.
+ final Map<String, Map<String, ConnectorValueReference>>
migratedActiveProperties = migrateProperties(activeConfig);
+ final Map<String, Map<String, ConnectorValueReference>>
migratedWorkingProperties = migrateProperties(workingConfig);
+
+ final MutableConnectorConfigurationContext activeSeedContext =
createConfigurationContext(migratedActiveProperties);
final FrameworkFlowContext inheritContext =
flowContextFactory.createWorkingFlowContext(identifier,
- connectorDetails.getComponentLog(), configurationContext,
flowContextBundle);
+ connectorDetails.getComponentLog(), activeSeedContext,
flowContextBundle);
- // Apply the update for the active config
+ // Apply the active configuration. This restores activeFlowContext to
migratedActiveProperties and internally
+ // rebuilds workingFlowContext aliased to activeFlowContext's
configuration; we discard that alias below and
+ // construct an independent working context so active and working do
not share configuration state when the
+ // two lists actually diverge.
applyUpdate(inheritContext);
- // Configure the working config but do not apply
- for (final VersionedConfigurationStep step : workingConfig) {
- final StepConfiguration stepConfig = createStepConfiguration(step);
- setConfiguration(step.getName(), stepConfig, true);
+ // Tear down the working context that applyUpdate created aliased to
active, and rebuild it around an
+ // independent configuration seeded from migratedWorkingProperties.
Then fire onConfigurationStepConfigured
+ // for every step so renamed steps trigger the flow-builder callback
under their new name and any
+ // value-derived flow state (resolved asset paths, secret values,
etc.) is populated against the fresh
+ // working context.
+ destroyWorkingContext();
+ final MutableConnectorConfigurationContext workingConfigContext =
createConfigurationContext(migratedWorkingProperties);
+ workingFlowContext =
flowContextFactory.createWorkingFlowContext(identifier,
+ connectorDetails.getComponentLog(), workingConfigContext,
flowContextBundle);
Review Comment:
Not a big deal but hygiene-wise it's best to avoid line wraps for short
lines like this.
##########
nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorPropertyMigrationIT.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.tests.system.connectors;
+
+import org.apache.nifi.tests.system.AbstractNarSwapMigrationIT;
+import org.apache.nifi.toolkit.client.NiFiClientException;
+import org.apache.nifi.web.api.dto.ConfigurationStepConfigurationDTO;
+import org.apache.nifi.web.api.dto.ConnectorConfigurationDTO;
+import org.apache.nifi.web.api.dto.ConnectorValueReferenceDTO;
+import org.apache.nifi.web.api.dto.PropertyGroupConfigurationDTO;
+import org.apache.nifi.web.api.entity.ConnectorEntity;
+import org.apache.nifi.web.api.entity.ParameterProviderEntity;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * End-to-end system test for {@code Connector.migrateProperties}. A
pre-migration
+ * {@code MigratePropertiesConnector} fixture in {@code
nifi-system-test-extensions-nar} is created and configured
+ * through the REST API using legacy step / property names. NiFi is then
stopped, the system-test extensions NARs are
+ * swapped for {@code nifi-alternate-config-extensions-nar} (which contains a
same-simple-typed-name fixture that
+ * implements {@code migrateProperties}), and NiFi is restarted. The
framework's {@code StandardConnectorRepository}
+ * sync path invokes {@code StandardConnectorNode.inheritConfiguration}, which
drives
+ * {@code Connector.migrateProperties}. The test asserts that the persisted
active and working configurations both
+ * reflect the migrated step and property names / values.
+ */
+public class ConnectorPropertyMigrationIT extends AbstractNarSwapMigrationIT {
+
+ private static final String CONNECTOR_TYPE_SIMPLE_NAME =
"MigratePropertiesConnector";
+
+ private static final String LEGACY_STEP_NAME = "Legacy Step";
+ private static final String LEGACY_BROKER_URL_NAME = "legacy-broker-url";
+ private static final String LEGACY_CREDENTIALS_NAME = "legacy-credentials";
+ private static final String LEGACY_OBSOLETE_NAME = "legacy-obsolete";
+
+ private static final String MIGRATED_STEP_NAME = "Kafka Connection";
+ private static final String BROKER_URL_NAME = "Broker URL";
+ private static final String CREDENTIALS_NAME = "Credentials";
+ private static final String CLIENT_ID_NAME = "Client Id";
+
+ private static final String BROKER_URL_VALUE = "kafka-broker:9092";
+ private static final String OBSOLETE_VALUE = "gone";
+ private static final String CLIENT_ID_MIGRATED_VALUE = "auto-migrated";
+
+ private static final String SECRET_NAME = "supersecret";
+ private static final String FULLY_QUALIFIED_SECRET_NAME =
"PropertiesParameterProvider.Parameters.supersecret";
+ private static final String STRING_LITERAL_TYPE = "STRING_LITERAL";
+ private static final String SECRET_REFERENCE_TYPE = "SECRET_REFERENCE";
+
+ @Test
+ public void testConnectorPropertyMigration() throws NiFiClientException,
IOException, InterruptedException {
+ // Stand up a Parameter Provider with a single secret so that the
legacy secret property can be configured with
+ // a SECRET_REFERENCE that survives the migration.
+ final ParameterProviderEntity paramProvider =
getClientUtil().createParameterProvider("PropertiesParameterProvider");
+ getClientUtil().updateParameterProviderProperties(paramProvider,
Map.of("parameters", SECRET_NAME + "=" + SECRET_NAME));
+
+ // Create the pre-migration connector and configure the legacy step
with a string literal, a secret reference,
+ // and an obsolete string literal that will be removed by migration.
+ final ConnectorEntity connector =
getClientUtil().createConnector(CONNECTOR_TYPE_SIMPLE_NAME);
+ assertNotNull(connector);
+
+ final ConnectorValueReferenceDTO brokerUrlRef =
createStringLiteralReference(BROKER_URL_VALUE);
+ final ConnectorValueReferenceDTO credentialsRef =
getClientUtil().createSecretValueReference(paramProvider.getId(),
+ SECRET_NAME, FULLY_QUALIFIED_SECRET_NAME);
+ final ConnectorValueReferenceDTO obsoleteRef =
createStringLiteralReference(OBSOLETE_VALUE);
+ final Map<String, ConnectorValueReferenceDTO> legacyProperties =
Map.of(
+ LEGACY_BROKER_URL_NAME, brokerUrlRef,
+ LEGACY_CREDENTIALS_NAME, credentialsRef,
+ LEGACY_OBSOLETE_NAME, obsoleteRef
+ );
+ getClientUtil().configureConnectorWithReferences(connector.getId(),
LEGACY_STEP_NAME, legacyProperties);
+ getClientUtil().applyConnectorUpdate(connector);
+
+ // Sanity check: pre-restart state still has the legacy schema before
we swap NARs.
+ final ConnectorEntity beforeRestart =
getNifiClient().getConnectorClient().getConnector(connector.getId());
+ final Map<String, ConnectorValueReferenceDTO> legacyActive =
propertyValues(beforeRestart.getComponent().getActiveConfiguration(),
LEGACY_STEP_NAME);
+ assertEquals(BROKER_URL_VALUE,
legacyActive.get(LEGACY_BROKER_URL_NAME).getValue());
+ assertEquals(SECRET_REFERENCE_TYPE,
legacyActive.get(LEGACY_CREDENTIALS_NAME).getValueType());
+ assertEquals(OBSOLETE_VALUE,
legacyActive.get(LEGACY_OBSOLETE_NAME).getValue());
+
+ // Swap out the system-test-extensions NARs for the alternate-config
NAR that defines the same simple-typed
+ // MigratePropertiesConnector with the migrated schema and a
migrateProperties implementation.
+ getNiFiInstance().stop();
+ switchOutNars();
+ getNiFiInstance().start(true);
+
+ // On restart, StandardConnectorRepository.syncConnector ->
StandardConnectorNode.inheritConfiguration invokes
+ // Connector.migrateProperties for both the active and working
configurations. Assert that each context now
+ // reflects the renamed step, renamed / removed / added properties,
and that typed value references survive.
+ final ConnectorEntity afterRestart =
getNifiClient().getConnectorClient().getConnector(connector.getId());
+
assertMigratedConfiguration(afterRestart.getComponent().getActiveConfiguration(),
"active");
+
assertMigratedConfiguration(afterRestart.getComponent().getWorkingConfiguration(),
"working");
+
+ getClientUtil().waitForValidConnector(connector.getId());
+ }
+
+ private void assertMigratedConfiguration(final ConnectorConfigurationDTO
config, final String label) {
+ final List<ConfigurationStepConfigurationDTO> steps =
config.getConfigurationStepConfigurations();
+ assertNotNull(steps, label + " configuration is missing steps");
+ assertTrue(steps.stream().noneMatch(step ->
LEGACY_STEP_NAME.equals(step.getConfigurationStepName())),
+ label + " configuration still contains legacy step");
+
+ final Map<String, ConnectorValueReferenceDTO> migrated =
propertyValues(config, MIGRATED_STEP_NAME);
+
+ final ConnectorValueReferenceDTO brokerUrl =
migrated.get(BROKER_URL_NAME);
+ assertNotNull(brokerUrl, label + " configuration is missing " +
BROKER_URL_NAME);
+ assertEquals(STRING_LITERAL_TYPE, brokerUrl.getValueType());
+ assertEquals(BROKER_URL_VALUE, brokerUrl.getValue());
+
+ final ConnectorValueReferenceDTO credentials =
migrated.get(CREDENTIALS_NAME);
+ assertNotNull(credentials, label + " configuration is missing " +
CREDENTIALS_NAME);
+ assertEquals(SECRET_REFERENCE_TYPE, credentials.getValueType());
+ assertEquals(SECRET_NAME, credentials.getSecretName());
+
+ final ConnectorValueReferenceDTO clientId =
migrated.get(CLIENT_ID_NAME);
+ assertNotNull(clientId, label + " configuration is missing " +
CLIENT_ID_NAME);
+ assertEquals(STRING_LITERAL_TYPE, clientId.getValueType());
+ assertEquals(CLIENT_ID_MIGRATED_VALUE, clientId.getValue());
+
+ assertFalse(migrated.containsKey(LEGACY_OBSOLETE_NAME),
+ label + " configuration still contains obsolete legacy property");
+ assertFalse(migrated.containsKey(LEGACY_BROKER_URL_NAME),
+ label + " configuration still contains legacy broker URL name");
+ assertFalse(migrated.containsKey(LEGACY_CREDENTIALS_NAME),
+ label + " configuration still contains legacy credentials name");
Review Comment:
Another example of overly aggressive line wraps - and making unnecessary
assertions where the `assertFalse` already makes it clear that it shouldn't be
there but is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
