InternalTopicManager

InternalTopicManager is created exclusively for the StreamsPartitionAssignor so it can make repartition and state changelog internal topics ready.

InternalTopicManager uses a Kafka AdminClient to create and describe topics.

InternalTopicManager uses stream-thread [threadName] for the logContext.

Table 1. InternalTopicManager’s Internal Properties (e.g. Registries, Counters and Flags)
Name	Description
`defaultTopicConfigs`	Default topic configurations (`Map<String, String>`) that is filled out with new entries (with `topic.` prefix) when `InternalTopicManager` is created
`log`	`Logger`
`replicationFactor`	Value of replication.factor configuration property
`retries`	Value of AdminClient’s `retries` configuration property (to resend any request that fails with a potentially transient error)
`windowChangeLogAdditionalRetention`	Value of windowstore.changelog.additional.retention.ms configuration property

Tip

Enable any of DEBUG logging level for org.apache.kafka.streams.processor.internals.InternalTopicManager logger to see what happens inside.

Add the following line to log4j.properties:

log4j.logger.org.apache.kafka.streams.processor.internals.InternalTopicManager=DEBUG

Refer to Application Logging Using log4j.

Creating InternalTopicManager Instance

InternalTopicManager takes the following when created:

Kafka AdminClient
StreamsConfig

InternalTopicManager initializes the internal registries and counters.

While being created, InternalTopicManager prints out the following DEBUG message to the logs:

Configs:
  retries = [retries]
  replication.factor = [replicationFactor]
  windowstore.changelog.additional.retention.ms = [windowChangeLogAdditionalRetention]

In the end, InternalTopicManager requests the StreamsConfig for originalsWithPrefix with topic. prefix and adds every non-empty configuration to the defaultTopicConfigs.

`makeReady` Method

void makeReady(final Map<String, InternalTopicConfig> topics)

makeReady…FIXME

Note	`makeReady` is used exclusively when `StreamsPartitionAssignor` is requested to prepareTopic (when requested to perform group assignment).

`getNumPartitions` Method

Map<String, Integer> getNumPartitions(final Set<String> topics)

getNumPartitions…FIXME

Note	`getNumPartitions` is used when `InternalTopicManager` is requested to makeReady.

InternalTopicManager