KAFKA-10000: Zombie fencing (KIP-618)

checkstyle/suppressions.xml

@@ -137,7 +137,7 @@
               files="ConfigKeyInfo.java"/>
 
     <suppress checks="ClassDataAbstractionCoupling"
-              files="(RestServer|AbstractHerder|DistributedHerder).java"/>
+              files="(RestServer|AbstractHerder|DistributedHerder|Worker).java"/>
 
     <suppress checks="BooleanExpressionComplexity"
               files="JsonConverter.java"/>

connect/basic-auth-extension/src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilter.java

@@ -18,7 +18,11 @@
 package org.apache.kafka.connect.rest.basic.auth.extension;
 
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
+import java.util.function.Predicate;
 import java.util.regex.Pattern;
 import javax.security.auth.login.Configuration;
 import javax.ws.rs.HttpMethod;
@@ -45,21 +49,40 @@
 public class JaasBasicAuthFilter implements ContainerRequestFilter {
 
     private static final Logger log = LoggerFactory.getLogger(JaasBasicAuthFilter.class);
-    private static final Pattern TASK_REQUEST_PATTERN = Pattern.compile("/?connectors/([^/]+)/tasks/?");
+    private static final Set<RequestMatcher> INTERNAL_REQUEST_MATCHERS = new HashSet<>(Arrays.asList(
+            new RequestMatcher(HttpMethod.POST, "/?connectors/([^/]+)/tasks/?"),
+            new RequestMatcher(HttpMethod.PUT, "/?connectors/[^/]+/fence/?")
+    ));
     private static final String CONNECT_LOGIN_MODULE = "KafkaConnect";
 
     static final String AUTHORIZATION = "Authorization";
 
     // Package-private for testing
     final Configuration configuration;
 
+    private static class RequestMatcher implements Predicate<ContainerRequestContext> {
+        private final String method;
+        private final Pattern path;
+
+        public RequestMatcher(String method, String path) {
+            this.method = method;
+            this.path = Pattern.compile(path);
+        }
+
+        @Override
+        public boolean test(ContainerRequestContext requestContext) {
+            return requestContext.getMethod().equals(method)
+                    && path.matcher(requestContext.getUriInfo().getPath()).matches();
+        }
+    }
+
     public JaasBasicAuthFilter(Configuration configuration) {
         this.configuration = configuration;
     }
 
     @Override
     public void filter(ContainerRequestContext requestContext) throws IOException {
-        if (isInternalTaskConfigRequest(requestContext)) {
+        if (isInternalRequest(requestContext)) {
             log.trace("Skipping authentication for internal request");
             return;
         }
@@ -82,12 +105,10 @@ public void filter(ContainerRequestContext requestContext) throws IOException {
         }
     }
 
-    private static boolean isInternalTaskConfigRequest(ContainerRequestContext requestContext) {
-        return requestContext.getMethod().equals(HttpMethod.POST)
-            && TASK_REQUEST_PATTERN.matcher(requestContext.getUriInfo().getPath()).matches();
+    private boolean isInternalRequest(ContainerRequestContext requestContext) {
+        return INTERNAL_REQUEST_MATCHERS.stream().anyMatch(m -> m.test(requestContext));
     }
 
-
     public static class BasicAuthCallBackHandler implements CallbackHandler {
 
         private static final String BASIC = "basic";

connect/basic-auth-extension/src/test/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilterTest.java

@@ -42,8 +42,10 @@
 
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 
 public class JaasBasicAuthFilterTest {
@@ -58,7 +60,7 @@ public void testSuccess() throws IOException {
         ContainerRequestContext requestContext = setMock("Basic", "user", "password");
         jaasBasicAuthFilter.filter(requestContext);
 
-        verify(requestContext).getMethod();
+        verify(requestContext, atLeastOnce()).getMethod();
         verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
     }
 
@@ -69,7 +71,7 @@ public void testEmptyCredentialsFile() throws IOException {
         ContainerRequestContext requestContext = setMock("Basic", "user", "password");
         jaasBasicAuthFilter.filter(requestContext);
 
-        verify(requestContext).getMethod();
+        verify(requestContext, atLeastOnce()).getMethod();
         verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
     }
 
@@ -81,7 +83,7 @@ public void testBadCredential() throws IOException {
         jaasBasicAuthFilter.filter(requestContext);
 
         verify(requestContext).abortWith(any(Response.class));
-        verify(requestContext).getMethod();
+        verify(requestContext, atLeastOnce()).getMethod();
         verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
     }
 
@@ -93,7 +95,7 @@ public void testBadPassword() throws IOException {
         jaasBasicAuthFilter.filter(requestContext);
 
         verify(requestContext).abortWith(any(Response.class));
-        verify(requestContext).getMethod();
+        verify(requestContext, atLeastOnce()).getMethod();
         verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
     }
 
@@ -105,7 +107,7 @@ public void testUnknownBearer() throws IOException {
         jaasBasicAuthFilter.filter(requestContext);
 
         verify(requestContext).abortWith(any(Response.class));
-        verify(requestContext).getMethod();
+        verify(requestContext, atLeastOnce()).getMethod();
         verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
     }
 
@@ -117,7 +119,7 @@ public void testUnknownLoginModule() throws IOException {
         jaasBasicAuthFilter.filter(requestContext);
 
         verify(requestContext).abortWith(any(Response.class));
-        verify(requestContext).getMethod();
+        verify(requestContext, atLeastOnce()).getMethod();
         verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
     }
 
@@ -128,7 +130,7 @@ public void testUnknownCredentialsFile() throws IOException {
         jaasBasicAuthFilter.filter(requestContext);
 
         verify(requestContext).abortWith(any(Response.class));
-        verify(requestContext).getMethod();
+        verify(requestContext, atLeastOnce()).getMethod();
         verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
     }
 
@@ -139,17 +141,26 @@ public void testNoFileOption() throws IOException {
         jaasBasicAuthFilter.filter(requestContext);
 
         verify(requestContext).abortWith(any(Response.class));
-        verify(requestContext).getMethod();
+        verify(requestContext, atLeastOnce()).getMethod();
         verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
     }
 
     @Test
-    public void testPostWithoutAppropriateCredential() throws IOException {
+    public void testInternalTaskConfigEndpointSkipped() throws IOException {
+        testInternalEndpointSkipped(HttpMethod.POST, "connectors/connName/tasks");
+    }
+
+    @Test
+    public void testInternalZombieFencingEndpointSkipped() throws IOException {
+        testInternalEndpointSkipped(HttpMethod.PUT, "connectors/connName/fence");
+    }
+
+    private void testInternalEndpointSkipped(String method, String endpoint) throws IOException {
         UriInfo uriInfo = mock(UriInfo.class);
-        when(uriInfo.getPath()).thenReturn("connectors/connName/tasks");
+        when(uriInfo.getPath()).thenReturn(endpoint);
 
         ContainerRequestContext requestContext = mock(ContainerRequestContext.class);
-        when(requestContext.getMethod()).thenReturn(HttpMethod.POST);
+        when(requestContext.getMethod()).thenReturn(method);
         when(requestContext.getUriInfo()).thenReturn(uriInfo);
 
         File credentialFile = setupPropertyLoginFile(true);
@@ -158,8 +169,9 @@ public void testPostWithoutAppropriateCredential() throws IOException {
         jaasBasicAuthFilter.filter(requestContext);
 
         verify(uriInfo).getPath();
-        verify(requestContext).getMethod();
+        verify(requestContext, atLeastOnce()).getMethod();
         verify(requestContext).getUriInfo();
+        verifyNoMoreInteractions(requestContext);
     }
 
     @Test

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java

@@ -138,6 +138,18 @@ public interface Herder {
      */
     void putTaskConfigs(String connName, List<Map<String, String>> configs, Callback<Void> callback, InternalRequestSignature requestSignature);
 
+    /**
+     * Fence out any older task generations for a source connector, and then write a record to the config topic
+     * indicating that it is safe to bring up a new generation of tasks. If that record is already present, do nothing
+     * and invoke the callback successfully.
+     * @param connName the name of the connector to fence out, which must refer to a source connector; if the
+     *                 connector does not exist or is not a source connector, the callback will be invoked with an error
+     * @param callback callback to invoke upon completion
+     * @param requestSignature the signature of the request made for this connector;
+     *                         may be null if no signature was provided
+     */
+    void fenceZombieSourceTasks(String connName, Callback<Void> callback, InternalRequestSignature requestSignature);
+
     /**
      * Get a list of connectors currently running in this cluster.
      * @return A list of connector names