package com.ankurm.kafkabasics;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.ConfigDef;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.context.ActiveProfiles;

import java.lang.reflect.Field;
import java.util.List;
import java.util.Map;

import static org.assertj.core.api.Assertions.assertThat;

/**
 * Prints three columns for each setting that matters: the kafka-clients default, whatever Spring
 * Boot put on top of it, and therefore the effective value. Defaults move between releases and
 * this is cheaper than remembering which.
 *
 * <p>The client defaults are read out of {@code ProducerConfig}/{@code ConsumerConfig}'s own
 * {@code ConfigDef} by reflection, so they are the real ones for the kafka-clients version on
 * this classpath rather than the ones the documentation happened to describe.
 *
 * @see <a href="../../../../../docs/06-acknowledgement.md">docs/06-acknowledgement.md</a>
 */
@SpringBootTest
@ActiveProfiles("test")
@EmbeddedKafka(topics = { "orders" }, partitions = 3)
class EffectiveConfigTest {

	@Autowired
	ProducerFactory<String, OrderEvent> producerFactory;

	@Autowired
	ConsumerFactory<String, OrderEvent> consumerFactory;

	@Autowired
	KafkaListenerEndpointRegistry registry;

	private static final List<String> PRODUCER_KEYS = List.of(ProducerConfig.ACKS_CONFIG,
			ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, ProducerConfig.RETRIES_CONFIG,
			ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, ProducerConfig.LINGER_MS_CONFIG,
			ProducerConfig.BATCH_SIZE_CONFIG, ProducerConfig.COMPRESSION_TYPE_CONFIG,
			ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG);

	private static final List<String> CONSUMER_KEYS = List.of(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
			ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, ConsumerConfig.MAX_POLL_RECORDS_CONFIG,
			ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,
			ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, ConsumerConfig.ISOLATION_LEVEL_CONFIG,
			ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG);

	@Test
	void printEffectiveConfiguration() throws Exception {
		table("producer", clientDefaults(ProducerConfig.class), PRODUCER_KEYS,
				this.producerFactory.getConfigurationProperties());
		System.out.println();
		table("consumer", clientDefaults(ConsumerConfig.class), CONSUMER_KEYS,
				this.consumerFactory.getConfigurationProperties());

		Map<String, Object> producer = this.producerFactory.getConfigurationProperties();
		Map<String, Object> consumer = this.consumerFactory.getConfigurationProperties();

		System.out.println();
		System.out.println("=== listener container ===");
		for (MessageListenerContainer container : this.registry.getListenerContainers()) {
			System.out.printf("%-46s %s%n", "ackMode", container.getContainerProperties().getAckMode());
			System.out.printf("%-46s %s%n", "groupId", container.getGroupId());
			System.out.printf("%-46s %s%n", "concurrency", container.getContainerProperties().getClientId());
		}

		// Boot sets NOTHING on the producer beyond serializers and bootstrap servers. Durability
		// therefore comes entirely from the kafka-clients defaults, which since Kafka 3.0 are
		// acks=all and enable.idempotence=true. Absent is not unsafe here - but it does mean a
		// property set in a Kafka 2.x-era runbook will change behaviour when you remove it.
		assertThat(producer).doesNotContainKeys(ProducerConfig.ACKS_CONFIG,
				ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, ProducerConfig.RETRIES_CONFIG);

		// The kafka-clients default for enable.auto.commit is TRUE, and Boot does not override it.
		// The container does - but NOT by touching this shared factory. ListenerConsumer's
		// determineAutoCommit checks whether the ConsumerFactory config contains the key and, when
		// it does not, calls setProperty("enable.auto.commit", "false") on the per-container
		// Properties handed to createConsumer. So the factory map never shows it, before or after
		// the containers start, and reading the factory to find out whether auto-commit is on
		// gives you the wrong answer.
		assertThat(defaultOf(ConsumerConfig.class, ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG))
				.isEqualTo(true);
		assertThat(this.consumerFactory.getConfigurationProperties())
				.doesNotContainKey(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);
		// And the public accessor on the factory agrees with the factory, not with reality:
		// isAutoCommit() reads the same absent key and falls back to the client default, so it
		// answers TRUE for a stock Boot 4.1 application in which no consumer auto-commits.
		assertThat(this.consumerFactory.isAutoCommit()).isTrue();

		// It DOES set isolation.level, which is the one place Boot has an opinion.
		assertThat(consumer).containsEntry(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_uncommitted");
	}

	private void table(String title, Map<String, Object> defaults, List<String> keys,
			Map<String, Object> spring) {
		System.out.println("=== " + title + " ===");
		System.out.printf("%-46s %-28s %-28s%n", "property", "kafka-clients default", "set by Spring Boot");
		System.out.println("-".repeat(104));
		for (String key : keys) {
			System.out.printf("%-46s %-28s %-28s%n", key, String.valueOf(defaults.get(key)),
					spring.containsKey(key) ? spring.get(key) : "-");
		}
	}

	private static Map<String, Object> clientDefaults(Class<?> configClass) throws Exception {
		Field field = configClass.getDeclaredField("CONFIG");
		field.setAccessible(true);
		ConfigDef configDef = (ConfigDef) field.get(null);
		return configDef.defaultValues();
	}

	private static Object defaultOf(Class<?> configClass, String key) throws Exception {
		return clientDefaults(configClass).get(key);
	}

}