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

Code before:
        // 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);
            
            final StreamsConfig streamsConfig = new StreamsConfig(props);
            // Trigger the warning by getting consumer configs
            streamsConfig.getMainConsumerConfigs("group", "client", 0);

            assertTrue(appender.getMessages().stream()
                .anyMatch(msg -> msg.contains("Unexpected user-specified consumer config 'allow.auto.create.topics' found") 
                    && msg.contains("User setting (true) will be ignored and the Streams default setting (false) will be used")),
                "Should log error when user tries to override allow.auto.create.topics");
        }
    }


    @Test
    public void shouldSupportNonPrefixedAdminConfigs() {
        props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 10);
        final StreamsConfig streamsConfig = new StreamsConfig(props);
        final Map<String, Object> configs = streamsConfig.getAdminConfigs(clientId);
        assertEquals(10, configs.get(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG));
    }


Review comment (refactor):
Should we also duplicate this test, using the three individual consumer prefix? (or just merge this test into the one from above?

Code after:
        // 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");
        }
    }

    @Test
    public void shouldNotAllowAutoCreateTopicsForConsumers_WithSpecificConsumerPrefixes() {
        // Test with specific prefixes for each consumer type
        props.put(StreamsConfig.mainConsumerPrefix(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");

        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 main.consumer.* override");

            // Restore consumer - verify override is ignored