gharris1727 commented on code in PR #13465:
URL: https://github.com/apache/kafka/pull/13465#discussion_r1163174096
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java:
##########
@@ -1247,6 +1254,224 @@ void sourceConnectorOffsets(String connName,
ConnectorOffsetBackingStore offsetS
});
}
+ /**
+ * Alter a connector's offsets.
+ *
+ * @param connName the name of the connector whose offsets are to be
altered
+ * @param offsets a mapping from partitions to offsets that need to be
overwritten
+ * @param connectorConfig the connector's configurations
+ *
+ * @return true if the connector plugin has implemented {@link
org.apache.kafka.connect.sink.SinkConnector#alterOffsets(Map, Map)}
+ * / {@link
org.apache.kafka.connect.source.SourceConnector#alterOffsets(Map, Map)} and it
returns true for the provided offsets,
+ * false otherwise
+ *
+ */
+ public boolean alterConnectorOffsets(String connName, Map<Map<String, ?>,
Map<String, ?>> offsets,
Review Comment:
nit: swap order of offsets and config to be consistent with the plugin api
##########
connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java:
##########
@@ -19,24 +19,25 @@
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.runtime.isolation.Plugins;
import org.apache.kafka.connect.runtime.rest.entities.ActiveTopicsInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConfigInfos;
-import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffsets;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorInfo;
+import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffsets;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
import org.apache.kafka.connect.runtime.rest.entities.ServerInfo;
import org.apache.kafka.connect.runtime.rest.errors.ConnectRestException;
+import org.eclipse.jetty.client.HttpClient;
Review Comment:
Using raw java io is certainly clunky, but was there some reason that you
couldn't make it work like the existing calls?
I think since this is a class used outside of this repo, we should be
careful about adding new dependencies.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java:
##########
@@ -1247,6 +1254,224 @@ void sourceConnectorOffsets(String connName,
ConnectorOffsetBackingStore offsetS
});
}
+ /**
+ * Alter a connector's offsets.
+ *
+ * @param connName the name of the connector whose offsets are to be
altered
+ * @param offsets a mapping from partitions to offsets that need to be
overwritten
+ * @param connectorConfig the connector's configurations
+ *
+ * @return true if the connector plugin has implemented {@link
org.apache.kafka.connect.sink.SinkConnector#alterOffsets(Map, Map)}
+ * / {@link
org.apache.kafka.connect.source.SourceConnector#alterOffsets(Map, Map)} and it
returns true for the provided offsets,
+ * false otherwise
+ *
+ */
+ public boolean alterConnectorOffsets(String connName, Map<Map<String, ?>,
Map<String, ?>> offsets,
+ Map<String, String> connectorConfig) {
+ String connectorClassOrAlias =
connectorConfig.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG);
+ ClassLoader connectorLoader =
plugins.connectorLoader(connectorClassOrAlias);
+ Connector connector;
+
+ try (LoaderSwap loaderSwap = plugins.withClassLoader(connectorLoader))
{
+ connector = plugins.newConnector(connectorClassOrAlias);
+ if (ConnectUtils.isSinkConnector(connector)) {
+ log.debug("Altering consumer group offsets for sink connector:
{}", connName);
+ return alterSinkConnectorOffsets(connName, connector,
connectorConfig, offsets);
+ } else {
+ log.debug("Altering offsets for source connector: {}",
connName);
+ return alterSourceConnectorOffsets(connName, connector,
connectorConfig, offsets);
+ }
+ }
+ }
+
+ /**
+ * Alter a sink connector's consumer group offsets.
+ * <p>
+ * Visible for testing.
+ * @param connName the name of the sink connector whose offsets are to be
altered
+ * @param connector an instance of the sink connector
+ * @param connectorConfig the sink connector's configuration
+ * @param offsets a mapping from topic partitions to offsets that need to
be overwritten
+ * @return true if the sink connector has implemented {@link
org.apache.kafka.connect.sink.SinkConnector#alterOffsets(Map, Map)}
+ * and it returns true for the provided offsets, false otherwise
+ */
+ boolean alterSinkConnectorOffsets(String connName, Connector connector,
Map<String, String> connectorConfig,
+ Map<Map<String, ?>, Map<String, ?>>
offsets) {
+ Map<TopicPartition, Long> parsedOffsets =
SinkUtils.validateAndParseSinkConnectorOffsets(offsets);
+ Timer timer = time.timer(ALTER_OFFSETS_TIMEOUT_MS);
+ boolean alterOffsetsResult;
+ try {
+ alterOffsetsResult = ((SinkConnector)
connector).alterOffsets(connectorConfig, parsedOffsets);
+ } catch (UnsupportedOperationException e) {
+ throw new ConnectException("Failed to alter offsets for connector
" + connName + " because it doesn't support external " +
+ "modification of offsets", e);
+ }
+ timer.update();
Review Comment:
Should alterOffsets exceeding the timer cause us to bail out here, or should
the timeout only apply to the adminclient operation?
--
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]
