gharris1727 commented on code in PR #17988:
URL: https://github.com/apache/kafka/pull/17988#discussion_r1868558024
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/RequiredPluginsMetadata.java:
##########
@@ -0,0 +1,133 @@
+package org.apache.kafka.connect.runtime;
+
+import org.apache.kafka.connect.components.Versioned;
+import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.connector.Connector;
+import org.apache.kafka.connect.connector.Task;
+import org.apache.kafka.connect.health.ConnectorType;
+import org.apache.kafka.connect.runtime.isolation.LoaderSwap;
+import org.apache.kafka.connect.runtime.isolation.Plugins;
+import org.apache.kafka.connect.sink.SinkConnector;
+import org.apache.kafka.connect.source.SourceConnector;
+import org.apache.kafka.connect.storage.Converter;
+import org.apache.kafka.connect.storage.HeaderConverter;
+import org.apache.kafka.connect.transforms.Transformation;
+import org.apache.kafka.connect.util.PluginVersionUtils;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class RequiredPluginsMetadata {
+
+ private final String connectorClass;
+ private final String connectorVersion;
+ private final ConnectorType connectorType;
+ private final String taskClass;
+ private final String taskVersion;
+ private final String keyConverterClass;
+ private final String keyConverterVersion;
+ private final String valueConverterClass;
+ private final String valueConverterVersion;
+ private final String headerConverterClass;
+ private final String headerConverterVersion;
+ private final Set<TransformationStage.AliasedPluginInfo> transformations;
+ private final Set<TransformationStage.AliasedPluginInfo> predicates;
+
+ public RequiredPluginsMetadata(
+ Connector connector,
+ Task task,
+ Converter keyConverter,
+ Converter valueConverter,
+ HeaderConverter headerConverter,
+ List<TransformationStage.StageInfo> transformationStageInfo
+ ) {
+
+ assert connector != null;
+ assert task != null;
+ assert keyConverter != null;
+ assert valueConverter != null;
+ assert headerConverter != null;
+ assert transformationStageInfo != null;
+
+ this.connectorClass = connector.getClass().getName();
+ this.connectorVersion =
PluginVersionUtils.getVersionOrUndefined(connector);
+ this.connectorType = getConnectorType(connector);
+ this.taskClass = task.getClass().getName();
+ this.taskVersion = task.version();
+ this.keyConverterClass = keyConverter.getClass().getName();
+ this.keyConverterVersion =
PluginVersionUtils.getVersionOrUndefined(keyConverter);
+ this.valueConverterClass = valueConverter.getClass().getName();
+ this.valueConverterVersion =
PluginVersionUtils.getVersionOrUndefined(valueConverter);
+ this.headerConverterClass = headerConverter.getClass().getName();
+ this.headerConverterVersion =
PluginVersionUtils.getVersionOrUndefined(headerConverter);
+ this.transformations =
transformationStageInfo.stream().map(TransformationStage.StageInfo::transform).collect(Collectors.toSet());
+ this.predicates =
transformationStageInfo.stream().map(TransformationStage.StageInfo::predicate).filter(Objects::nonNull).collect(Collectors.toSet());
+ }
+
+
+ public ConnectorType getConnectorType(Connector connector) {
+ try (LoaderSwap swap =
Plugins.swapLoader(connector.getClass().getClassLoader())) {
Review Comment:
`instanceof` doesn't run any plugin code, and so doesn't need a classloader
swap.
--
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
