yashmayya commented on code in PR #13465:
URL: https://github.com/apache/kafka/pull/13465#discussion_r1163144352
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java:
##########
@@ -1247,6 +1261,229 @@ 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.
+ * @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
+ */
+ private boolean alterSinkConnectorOffsets(String connName, Connector
connector, Map<String, String> connectorConfig,
+ Map<Map<String, ?>, Map<String,
?>> offsets) {
+ return alterSinkConnectorOffsets(connName, connector, connectorConfig,
offsets, Admin::create);
+ }
+
+ // Visible for testing; allows mocking the admin client for testing
Review Comment:
Ha, yeah, I really liked the pattern!
> It'd be more succinct if we added a new Worker constructor that allows us
to specify an admin factory instead of new methods.
Thanks, great suggestion, done. The verbosity and the number of superfluous
methods was bothering me too, this seems much better now.
--
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]
