psiby commented on code in PR #23268:
URL: https://github.com/apache/kafka/pull/23268#discussion_r4229187647
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java:
##########
@@ -1316,6 +1316,66 @@ public void resetConnectorOffsets(String connName,
Callback<Message> callback) {
*/
protected abstract void modifyConnectorOffsets(String connName,
Map<Map<String, ?>, Map<String, ?>> offsets, Callback<Message> cb);
+ /**
+ * Validate the offsets supplied via the {@code initial_offsets} field of
a connector creation request. Performs no
+ * I/O, so that an invalid request is rejected before any of the
connector's existing offsets are wiped.
+ *
+ * @param connName the name of the connector being created
+ * @param initialOffsets the offsets supplied by the user; may not be
{@code null}
+ * @throws BadRequestException if the offsets cannot be used as a
connector's initial offsets
+ */
+ protected void validateInitialOffsets(String connName, Map<Map<String, ?>,
Map<String, ?>> initialOffsets) {
+ if (initialOffsets.isEmpty()) {
+ throw new BadRequestException("The initial offsets for connector "
+ connName + " may not be empty");
+ }
+ // A null offset means "delete this partition's offset" when altering,
but is meaningless on create: every offset
+ // the connector has is wiped before these are written, so there is
nothing left to delete. Issuing the delete
+ // anyway targets the consumer group our own wipe just removed, which
the alter path (unlike the reset path) does
+ // not tolerate. This is deliberately stricter than PATCH
/connectors/{connector}/offsets.
+ // Not Map::containsValue, which throws rather than returning false on
the immutable maps Map.of(...) produces.
+ if (initialOffsets.values().stream().anyMatch(offset -> offset ==
null)) {
+ throw new BadRequestException("The initial offsets for connector "
+ connName + " may not contain null offset "
+ + "values. All existing offsets for the connector are
removed before the initial offsets are written, "
+ + "so a null offset has nothing to delete.");
+ }
+ }
+
+ /**
+ * Wipe every offset the connector currently has and write {@code
initialOffsets} in their place.
+ * <p>
+ * Unlike {@link #alterConnectorOffsets} and {@link
#resetConnectorOffsets}, this skips the usual preconditions (that
+ * the connector exists and is STOPPED with no running tasks): the
connector is being created, so it is not expected
+ * to exist and has no tasks. Its config has not been written yet, so
nothing can be running under this name.
+ */
+ protected void setInitialConnectorOffsets(String connName, Map<String,
String> config,
+ Map<Map<String, ?>, Map<String,
?>> initialOffsets, Callback<Message> cb) {
+ worker.modifyConnectorOffsets(connName, config, initialOffsets, true,
cb);
+ }
+
+ /**
+ * Remove the initial offsets written for a connector whose config could
not subsequently be written, so that a
+ * failed creation does not leave offsets behind for a connector that does
not exist, and then complete
+ * {@code callback} with the original failure rather than with any error
from the cleanup itself.
+ */
+ protected void wipeInitialOffsetsAfterFailedCreate(String connName,
Map<String, String> config,
+ Throwable
configWriteError, Callback<Created<ConnectorInfo>> callback) {
+ log.error("Failed to write the configuration for connector {} after
its initial offsets had been written; "
Review Comment:
Good point, changed it to "Failed to create connector ..." and updated the
javadoc above it to match.
--
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]
