C0urante commented on code in PR #12789:
URL: https://github.com/apache/kafka/pull/12789#discussion_r1008680032
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java:
##########
@@ -686,21 +690,25 @@ protected Connector getConnector(String connType) {
return tempConnectors.computeIfAbsent(connType, k ->
plugins().newConnector(k));
}
- /**
- * Retrieves ConnectorType for the corresponding connector class
- * @param connClass class of the connector
- */
- public ConnectorType connectorTypeForClass(String connClass) {
- return ConnectorType.from(getConnector(connClass).getClass());
- }
-
/**
* Retrieves ConnectorType for the class specified in the connector config
- * @param connConfig the connector config; may not be null
- * @return the {@link ConnectorType} of the connector
+ * @param connConfig the connector config, may be null
+ * @return the {@link ConnectorType} of the connector, or {@link
ConnectorType#UNKNOWN} if an error occurs
*/
- public ConnectorType connectorTypeForConfig(Map<String, String>
connConfig) {
- return
connectorTypeForClass(connConfig.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG));
+ public ConnectorType connectorType(Map<String, String> connConfig) {
+ if (connConfig == null) {
+ return ConnectorType.UNKNOWN;
+ }
+ String connClass =
connConfig.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG);
+ if (connClass == null) {
+ return ConnectorType.UNKNOWN;
+ }
+ try {
+ return ConnectorType.from(getConnector(connClass).getClass());
+ } catch (ConnectException e) {
+ log.error("Unable to retrieve connector type", e);
Review Comment:
Could we downgrade this to `WARN`? `ERROR` may be a bit too noisy,
especially if there's a UI or some other monitoring tool that's repeatedly
making requests to `GET /connectors?expand=status`.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java:
##########
@@ -686,21 +690,25 @@ protected Connector getConnector(String connType) {
return tempConnectors.computeIfAbsent(connType, k ->
plugins().newConnector(k));
}
- /**
- * Retrieves ConnectorType for the corresponding connector class
- * @param connClass class of the connector
- */
- public ConnectorType connectorTypeForClass(String connClass) {
- return ConnectorType.from(getConnector(connClass).getClass());
- }
-
/**
* Retrieves ConnectorType for the class specified in the connector config
- * @param connConfig the connector config; may not be null
- * @return the {@link ConnectorType} of the connector
+ * @param connConfig the connector config, may be null
+ * @return the {@link ConnectorType} of the connector, or {@link
ConnectorType#UNKNOWN} if an error occurs
Review Comment:
Nit:
```suggestion
* @return the {@link ConnectorType} of the connector, or {@link
ConnectorType#UNKNOWN} if an error occurs or the type cannot be determined
```
(Since we gracefully handle null input by returning `UNKNOWN`, we shouldn't
imply that it's only returned if an error occurs.)
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java:
##########
@@ -192,6 +193,41 @@ public void testConnectorStatus() {
ConnectorStateInfo state = herder.connectorStatus(connectorName);
assertEquals(connectorName, state.name());
+ assertEquals(ConnectorType.UNKNOWN, state.type());
+ assertEquals("RUNNING", state.connector().state());
+ assertEquals(1, state.tasks().size());
+ assertEquals(workerId, state.connector().workerId());
+
+ ConnectorStateInfo.TaskState taskState = state.tasks().get(0);
+ assertEquals(0, taskState.id());
+ assertEquals("UNASSIGNED", taskState.state());
+ assertEquals(workerId, taskState.workerId());
Review Comment:
I like these additional assertions on the accuracy of the
`ConnectorStateInfo` returned from `herder::connectorStatus`, but aren't we
testing a bit of an edge case here since we're simulating a status being
present for the connector but not a config?
I think if we want to add these assertions we should also tweak the test to
match the more common case where both a config and statuses can be found for
the connector.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerderTest.java:
##########
@@ -127,7 +127,7 @@ private enum SourceSink {
@Before
public void setup() {
worker = PowerMock.createMock(Worker.class);
- String[] methodNames = new String[]{"connectorTypeForClass"/*,
"validateConnectorConfig"*/, "buildRestartPlan", "recordRestarting"};
+ String[] methodNames = new String[]{"connectorType"/*,
"validateConnectorConfig"*/, "buildRestartPlan", "recordRestarting"};
Review Comment:
While we're in the neighborhood?
```suggestion
String[] methodNames = new String[]{"connectorType",
"buildRestartPlan", "recordRestarting"};
```
--
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]
