yashmayya commented on code in PR #13465: URL: https://github.com/apache/kafka/pull/13465#discussion_r1163225504
########## 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: Yeah, I did want to avoid doing this, but it was essentially a last resort. The `HttpURLConnection` API doesn't support the `PATCH` HTTP method and there is a won't fix bug ticket in OpenJDK for this - https://bugs.openjdk.org/browse/JDK-7016595. The new `HttpClient` (incubated in Java 9 and made official in Java 11) which supersedes `HttpURLConnection` does fix this issue, but since we still support Java 8, that wasn't an option either. Since the Jetty HTTP client is already being used by the main Connect codebase, I figured it'd be the best option here. ########## 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: Good point, I figured that in this case the next admin client operation would just use a timeout of 0 and timeout immediately. But now that I think about it, this'll lead to a misleading exception being thrown (indicating an issue with the admin client operation when the actual issue was the call to the plugin's `alterOffsets` method). We have a number of these timer updates after various operations in both the sink and source alter offsets implementations. While we can't use timeouts for these operations, maybe we can use a utility method which updates the timer and throws a `ConnectException` with a relevant message if the timer expires? WDYT? -- 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]
