MINOR: regression test for task assignor config
Reviewers: A. Sophie Blee-Goldman
@@ -27,6 +27,8 @@ import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
import org.apache.kafka.streams.processor.internals.StreamThread;
import org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor;
import org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration;
+import org.apache.kafka.streams.processor.internals.assignment.HighAvailabilityTaskAssignor;
+import org.apache.kafka.streams.processor.internals.assignment.TaskAssignor;
import org.apache.kafka.test.IntegrationTest;
import org.junit.ClassRule;
import org.junit.Rule;
@@ -38,12 +40,14 @@ import java.lang.reflect.Field;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.apache.kafka.common.utils.Utils.mkObjectProperties;
import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.sameInstance;
@@ -55,6 +59,9 @@ public class TaskAssignorIntegrationTest {
@Rule
public TestName testName = new TestName();
+ // Just a dummy implementation so we can check the config
+ public static final class MyTaskAssignor extends HighAvailabilityTaskAssignor implements TaskAssignor { }
+
@SuppressWarnings("unchecked")
@Test
public void shouldProperlyConfigureTheAssignor() throws NoSuchFieldException, IllegalAccessException {
@@ -88,8 +95,9 @@ public class TaskAssignorIntegrationTest {
mkEntry(StreamsConfig.ACCEPTABLE_RECOVERY_LAG_CONFIG, "6"),
mkEntry(StreamsConfig.MAX_WARMUP_REPLICAS_CONFIG, "7"),
mkEntry(StreamsConfig.PROBING_REBALANCE_INTERVAL_MS_CONFIG, "480000"),
+ mkEntry(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 9),
mkEntry(StreamsConfig.InternalConfig.ASSIGNMENT_LISTENER, configuredAssignmentListener),
- mkEntry(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 9)
+ mkEntry(StreamsConfig.InternalConfig.INTERNAL_TASK_ASSIGNOR_CLASS, MyTaskAssignor.class.getName())
)
);
@@ -127,12 +135,19 @@ public class TaskAssignorIntegrationTest {
final int adminClientTimeout =
(int) adminClientTimeoutField.get(streamsPartitionAssignor);
+ final Field taskAssignorSupplierField = StreamsPartitionAssignor.class.getDeclaredField("taskAssignorSupplier");
+ taskAssignorSupplierField.setAccessible(true);
+ final Supplier<TaskAssignor> taskAssignorSupplier =
+ (Supplier<TaskAssignor>) taskAssignorSupplierField.get(streamsPartitionAssignor);
+ final TaskAssignor taskAssignor = taskAssignorSupplier.get();
+
assertThat(configs.numStandbyReplicas, is(5));
assertThat(configs.acceptableRecoveryLag, is(6L));
assertThat(configs.maxWarmupReplicas, is(7));
assertThat(configs.probingRebalanceIntervalMs, is(480000L));
assertThat(actualAssignmentListener, sameInstance(configuredAssignmentListener));
assertThat(adminClientTimeout, is(9));
+ assertThat(taskAssignor, instanceOf(MyTaskAssignor.class));
}
}
}
