Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java (Java)

Code before:
        props.put(producerPrefix("custom.property.host"), "host2");
        props.put(adminClientPrefix("custom.property.host"), "host3");
        final StreamsConfig streamsConfig = new StreamsConfig(props);
        final Map<String, Object> consumerConfigs = streamsConfig.getMainConsumerConfigs(groupId, clientId, threadIdx);
        final Map<String, Object> restoreConsumerConfigs = streamsConfig.getRestoreConsumerConfigs(clientId);
        final Map<String, Object> producerConfigs = streamsConfig.getProducerConfigs(clientId);
        final Map<String, Object> adminConfigs = streamsConfig.getAdminConfigs(clientId);
        assertEquals("host1", consumerConfigs.get("custom.property.host"));
        assertEquals("host1", restoreConsumerConfigs.get("custom.property.host"));
        assertEquals("host2", producerConfigs.get("custom.property.host"));
        assertEquals("host3", adminConfigs.get("custom.property.host"));
    }

    @Test
    public void shouldOverrideAdminDefaultAdminClientEnableTelemetry() {
        final Map<String, Object> returnedProps = streamsConfig.getAdminConfigs(clientId);
        assertTrue((boolean) returnedProps.get(AdminClientConfig.ENABLE_METRICS_PUSH_CONFIG));
    }

    @Test
    public void testAutoCreateTopicsCannotBeOverriddenForStreamsConsumers() {
        // User tries to override the setting
        props.put(StreamsConfig.consumerPrefix(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), "true");
        props.put(StreamsConfig.restoreConsumerPrefix(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), "true");
        props.put(StreamsConfig.globalConsumerPrefix(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), "true");
        props.put(StreamsConfig.mainConsumerPrefix(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), "true");

        final StreamsConfig streamsConfig = new StreamsConfig(props);

        // Main consumer - verify override is ignored
        final Map<String, Object> mainConfigs = streamsConfig.getMainConsumerConfigs("group", "client", 0);
        assertEquals("false", mainConfigs.get(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG),
                "Main consumer should not allow auto topic creation even with override");

        // Restore consumer - verify override is ignored
        final Map<String, Object> restoreConfigs = streamsConfig.getRestoreConsumerConfigs("client");
        assertEquals("false", restoreConfigs.get(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG),
                "Restore consumer should not allow auto topic creation even with override");

        // Global consumer - verify override is ignored
        final Map<String, Object> globalConfigs = streamsConfig.getGlobalConsumerConfigs("client");
        assertEquals("false", globalConfigs.get(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG),
                "Global consumer should not allow auto topic creation even with override");
    }

    @Test
    public void shouldLogErrorWhenUserTriesToOverrideAutoCreateTopics() {
        props.put(StreamsConfig.consumerPrefix(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), "true");

        try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
            appender.setClassLogger(StreamsConfig.class, Level.ERROR);

Review comment (bug):
It seems we do not really test both prefix with this test setup. Because we set both the generic `consumer.` and the individual consumer prefix, the verification below will always fail on the former, and we never reach the code to fail on the later.

Guess we need to split into two test, one test using only `consumer.` (which set it's on all three cosumers), and one with set it's three time for each consumer indivivually. 

Code after:
        props.put(producerPrefix("custom.property.host"), "host2");
        props.put(adminClientPrefix("custom.property.host"), "host3");
        final StreamsConfig streamsConfig = new StreamsConfig(props);
        final Map<String, Object> consumerConfigs = streamsConfig.getMainConsumerConfigs(groupId, clientId, threadIdx);
        final Map<String, Object> restoreConsumerConfigs = streamsConfig.getRestoreConsumerConfigs(clientId);
        final Map<String, Object> producerConfigs = streamsConfig.getProducerConfigs(clientId);
        final Map<String, Object> adminConfigs = streamsConfig.getAdminConfigs(clientId);
        assertEquals("host1", consumerConfigs.get("custom.property.host"));
        assertEquals("host1", restoreConsumerConfigs.get("custom.property.host"));
        assertEquals("host2", producerConfigs.get("custom.property.host"));
        assertEquals("host3", adminConfigs.get("custom.property.host"));
    }

    @Test
    public void shouldOverrideAdminDefaultAdminClientEnableTelemetry() {
        final Map<String, Object> returnedProps = streamsConfig.getAdminConfigs(clientId);
        assertTrue((boolean) returnedProps.get(AdminClientConfig.ENABLE_METRICS_PUSH_CONFIG));
    }

    @Test
    public void shouldNotAllowAutoCreateTopicsForConsumers_WithCommonConsumerPrefix() {
        // Test with generic consumer.* prefix (affects all consumer types)
        props.put(StreamsConfig.consumerPrefix(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), "true");

        try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
            appender.setClassLogger(StreamsConfig.class, Level.ERROR);

            final StreamsConfig streamsConfig = new StreamsConfig(props);

            // Main consumer - verify override is ignored
            final Map<String, Object> mainConfigs = streamsConfig.getMainConsumerConfigs("group", "client", 0);
            assertEquals("false", mainConfigs.get(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG),
                    "Main consumer should not allow auto topic creation with consumer.* override");

            // Restore consumer - verify override is ignored
            final Map<String, Object> restoreConfigs = streamsConfig.getRestoreConsumerConfigs("client");
            assertEquals("false", restoreConfigs.get(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG),
                    "Restore consumer should not allow auto topic creation with consumer.* override");

            // Global consumer - verify override is ignored
            final Map<String, Object> globalConfigs = streamsConfig.getGlobalConsumerConfigs("client");
            assertEquals("false", globalConfigs.get(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG),
                    "Global consumer should not allow auto topic creation with consumer.* override");

            // Verify exactly 1 error is logged (consumer.* prefix is validated once in getCommonConsumerConfigs for each type of consumer)
            final List<String> errorMessages = appender.getMessages();
            final long errorCount = errorMessages.stream()
                    .filter(msg -> msg.contains("Unexpected user-specified consumer config 'allow.auto.create.topics' found"))
                    .count();
            assertEquals(3, errorCount,
                    "Should log exactly 3 error for consumer.* prefix");