yashmayya commented on code in PR #14704:
URL: https://github.com/apache/kafka/pull/14704#discussion_r1386195297
##########
docs/connect.html:
##########
@@ -41,7 +41,7 @@ <h4><a id="connect_running" href="#connect_running">Running
Kafka Connect</a></h
<p>In standalone mode all work is performed in a single process. This
configuration is simpler to setup and get started with and may be useful in
situations where only one worker makes sense (e.g. collecting log files), but
it does not benefit from some of the features of Kafka Connect such as fault
tolerance. You can start a standalone process with the following command:</p>
<pre class="brush: bash;">
-> bin/connect-standalone.sh config/connect-standalone.properties
[connector1.properties connector2.properties ...]</pre>
+> bin/connect-standalone.sh config/connect-standalone.properties
[connector1.properties connector2.json ...]</pre>
Review Comment:
Good point, I'd overlooked it because currently it only shows the worker
properties file and not the optional connector properties file (it's only
printed out if no argument is passed to the CLI). I've updated it to look
similar to this docs line.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/cli/ConnectStandalone.java:
##########
@@ -87,6 +98,61 @@ protected void processExtraArgs(Herder herder, Connect
connect, String[] extraAr
}
}
+ /**
+ * Parse a connector configuration file into a {@link
CreateConnectorRequest}. The file can have any one of the following formats:
+ * <ol>
+ * <li>A JSON file containing an Object with only String keys and
values that represent the connector configuration.</li>
+ * <li>A JSON file containing an Object that can be parsed directly
into a {@link CreateConnectorRequest}</li>
+ * <li>A valid Java Properties file (i.e. containing String key/value
pairs representing the connector configuration)</li>
+ * </ol>
+ * <p>
+ * Visible for testing.
+ *
+ * @param filePath the path of the connector configuration file
+ * @return the parsed connector configuration in the form of a {@link
CreateConnectorRequest}
+ */
+ CreateConnectorRequest parseConnectorConfigurationFile(String filePath)
throws IOException {
+ ObjectMapper objectMapper = new ObjectMapper();
+
+ File connectorConfigurationFile = Paths.get(filePath).toFile();
+ try {
+ Map<String, String> connectorConfigs = objectMapper.readValue(
+ connectorConfigurationFile,
+ new TypeReference<Map<String, String>>() { });
+
+ if (!connectorConfigs.containsKey(NAME_CONFIG)) {
+ throw new ConnectException("Connector configuration at '" +
filePath + "' is missing the mandatory '" + NAME_CONFIG + "' "
+ + "configuration");
+ }
+ return new
CreateConnectorRequest(connectorConfigs.get(NAME_CONFIG), connectorConfigs,
null);
+ } catch (StreamReadException | DatabindException e) {
+ log.debug("Could not parse connector configuration file '{}' into
a Map with String keys and values", filePath);
+ }
+
+ try {
+ CreateConnectorRequest createConnectorRequest =
objectMapper.readValue(connectorConfigurationFile,
+ new TypeReference<CreateConnectorRequest>() { });
+ if (createConnectorRequest.config().containsKey(NAME_CONFIG)) {
+ if
(!createConnectorRequest.config().get(NAME_CONFIG).equals(createConnectorRequest.name()))
{
+ throw new ConnectException("Connector name configuration
in 'config' doesn't match the one specified in 'name'");
+ }
+ } else {
+ createConnectorRequest.config().put(NAME_CONFIG,
createConnectorRequest.name());
+ }
+ return createConnectorRequest;
+ } catch (StreamReadException | DatabindException e) {
+ log.debug("Could not parse connector configuration file '{}' into
an object of type {}",
+ filePath, CreateConnectorRequest.class.getSimpleName());
+ }
+
+ Map<String, String> connectorConfigs =
Utils.propsToStringMap(Utils.loadProps(filePath));
Review Comment:
> IMO it'd be a bit friendlier to ignore unknown fields when attempting to
parse the file as a CreateConnectorRequest. Thoughts?
💯
I've changed this behavior and introduced an additional test, thanks for
pointing this out!
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java:
##########
@@ -108,6 +108,19 @@ public interface Herder {
*/
void putConnectorConfig(String connName, Map<String, String> config,
boolean allowReplace, Callback<Created<ConnectorInfo>> callback);
+ /**
+ * Set the configuration for a connector, along with a target state
optionally. This supports creation and updating.
+ * @param connName name of the connector
+ * @param config the connector's configuration
+ * @param targetState the desired target state for the connector; may be
{@code null} if no target state change is desired. Note that the default
+ * target state is {@link TargetState#STARTED} if no
target state exists previously
+ * @param allowReplace if true, allow overwriting previous configs; if
false, throw {@link AlreadyExistsException}
+ * if a connector with the same name already exists
+ * @param callback callback to invoke when the configuration has been
written
+ */
+ void putConnectorConfig(String connName, Map<String, String> config,
TargetState targetState, boolean allowReplace,
Review Comment:
I did ponder over this one for a while and the changes in the
`ConfigBackingStore` interface just introduce a new parameter to the existing
method instead of adding a new one (same situation as the `Herder` interface).
Here, however, I was a bit more hesitant to do the same because although it
isn't public API, I've seen some direct raw usages (not through the REST API)
of the `Herder` in the wild (i.e. outside of Connect). It's possible that the
direct usage in MM2 was used as an example / precedent to do so. It'd be nice
if we could avoid breaking such usages by introducing this new method instead,
WDYT?
##########
connect/runtime/src/main/java/org/apache/kafka/connect/cli/ConnectStandalone.java:
##########
@@ -87,6 +98,61 @@ protected void processExtraArgs(Herder herder, Connect
connect, String[] extraAr
}
}
+ /**
+ * Parse a connector configuration file into a {@link
CreateConnectorRequest}. The file can have any one of the following formats:
+ * <ol>
+ * <li>A JSON file containing an Object with only String keys and
values that represent the connector configuration.</li>
+ * <li>A JSON file containing an Object that can be parsed directly
into a {@link CreateConnectorRequest}</li>
+ * <li>A valid Java Properties file (i.e. containing String key/value
pairs representing the connector configuration)</li>
+ * </ol>
+ * <p>
+ * Visible for testing.
+ *
+ * @param filePath the path of the connector configuration file
+ * @return the parsed connector configuration in the form of a {@link
CreateConnectorRequest}
+ */
+ CreateConnectorRequest parseConnectorConfigurationFile(String filePath)
throws IOException {
+ ObjectMapper objectMapper = new ObjectMapper();
+
+ File connectorConfigurationFile = Paths.get(filePath).toFile();
+ try {
+ Map<String, String> connectorConfigs = objectMapper.readValue(
+ connectorConfigurationFile,
+ new TypeReference<Map<String, String>>() { });
+
+ if (!connectorConfigs.containsKey(NAME_CONFIG)) {
+ throw new ConnectException("Connector configuration at '" +
filePath + "' is missing the mandatory '" + NAME_CONFIG + "' "
+ + "configuration");
+ }
+ return new
CreateConnectorRequest(connectorConfigs.get(NAME_CONFIG), connectorConfigs,
null);
+ } catch (StreamReadException | DatabindException e) {
+ log.debug("Could not parse connector configuration file '{}' into
a Map with String keys and values", filePath);
+ }
+
+ try {
+ CreateConnectorRequest createConnectorRequest =
objectMapper.readValue(connectorConfigurationFile,
+ new TypeReference<CreateConnectorRequest>() { });
+ if (createConnectorRequest.config().containsKey(NAME_CONFIG)) {
+ if
(!createConnectorRequest.config().get(NAME_CONFIG).equals(createConnectorRequest.name()))
{
+ throw new ConnectException("Connector name configuration
in 'config' doesn't match the one specified in 'name'");
+ }
+ } else {
+ createConnectorRequest.config().put(NAME_CONFIG,
createConnectorRequest.name());
+ }
+ return createConnectorRequest;
+ } catch (StreamReadException | DatabindException e) {
+ log.debug("Could not parse connector configuration file '{}' into
an object of type {}",
+ filePath, CreateConnectorRequest.class.getSimpleName());
+ }
+
+ Map<String, String> connectorConfigs =
Utils.propsToStringMap(Utils.loadProps(filePath));
Review Comment:
> IMO it'd be a bit friendlier to ignore unknown fields when attempting to
parse the file as a CreateConnectorRequest. Thoughts?
💯
I've changed this behavior and introduced an additional test, thanks for
pointing this out!
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
