C0urante commented on code in PR #11779:
URL: https://github.com/apache/kafka/pull/11779#discussion_r891379608
##########
connect/basic-auth-extension/src/test/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilterTest.java:
##########
@@ -139,14 +140,23 @@ 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("connectors/connName/tasks");
+ }
+
+ @Test
+ public void testInternalZombieFencingEndpointSkipped() throws IOException {
+ testInternalEndpointSkipped("connectors/connName/fence");
+ }
+
+ private void testInternalEndpointSkipped(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);
Review Comment:
😱 this test was broken and did not catch calls to
`ContainerRequestContext::abortWith`. I've updated the test to catch those
calls and, after it started failing, also updated it to use the correct HTTP
method. Good catch, thanks!
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java:
##########
@@ -784,6 +845,14 @@ private static Map<String, Object>
connectorClientConfigOverrides(ConnectorTaskI
return clientOverrides;
}
+ private String transactionalId(ConnectorTaskId id) {
Review Comment:
Yep, exactly 👍
Going to try to do the rebase today, but may not be able to finish by EOD as
it's going to be fairly involved.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java:
##########
@@ -784,6 +845,14 @@ private static Map<String, Object>
connectorClientConfigOverrides(ConnectorTaskI
return clientOverrides;
}
+ private String transactionalId(ConnectorTaskId id) {
+ return transactionalId(config.groupId(), id.connector(), id.task());
+ }
+
+ public static String transactionalId(String groupId, String connector, int
taskId) {
Review Comment:
It's used in integration tests later on:
https://github.com/C0urante/kafka/blob/3d65e799925096d519b4adf906be05cba70addeb/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java#L828
##########
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java:
##########
@@ -201,6 +214,9 @@ public static String COMMIT_TASKS_KEY(String connectorName)
{
public static final Schema TARGET_STATE_V0 = SchemaBuilder.struct()
.field("state", Schema.STRING_SCHEMA)
.build();
+ public static final Schema TASK_COUNT_RECORD_V0 = SchemaBuilder.struct()
+ .field("tasks", Schema.INT32_SCHEMA)
Review Comment:
I think given the key format ("tasks-count-<connector>") this is
probably fine, and the name of the field is also specified in the KIP. But
similar to the 200 vs. 204 HTTP response for the fencing endpoint, this is
internal and a small detail, so I can change it if we agree that this kind of
detail doesn't need to precisely match what's in the KIP.
--
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]
