gharris1727 commented on code in PR #13465:
URL: https://github.com/apache/kafka/pull/13465#discussion_r1163322562
##########
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:
I don't think there's enough call-sites to justify a utility method, but
i'll leave that up to you.
--
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]
