Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@
import org.springframework.core.retry.RetryTemplate;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

/**
* Provide a wrapper API around the interactive query stores in Kafka Streams.
Expand All @@ -40,6 +39,7 @@
*
* @author Soby Chacko
* @author Stephane Nicoll
* @author Yingzi Zhang
*
* @since 3.2
*/
Expand Down Expand Up @@ -110,22 +110,23 @@ private void populateKafkaStreams() {

/**
* Retrieve the current {@link HostInfo} where this Kafka Streams application is running on.
* This {link @HostInfo} is different from the Kafka `bootstrap.server` property and is based on
* This {@link HostInfo} is different from the Kafka `bootstrap.server` property and is based on
* the Kafka Streams configuration property `application.server` where user-defined REST
* endpoints can be invoked per each Kafka Streams application instance.
* If this property - `application.server` - is not available from the end-user application, then null is returned.
* The value is parsed with {@link HostInfo#buildFromEndpoint(String)}, the same as Kafka Streams does,
* so IPv6 addresses such as {@code [::1]:8080} are supported.
* @return the current {@link HostInfo}
* @throws org.apache.kafka.common.config.ConfigException if `application.server` is not in the
* {@code host:port} format
*/
@Nullable
public HostInfo getCurrentKafkaStreamsApplicationHostInfo() {
Properties streamsConfiguration = this.streamsBuilderFactoryBean
.getStreamsConfiguration();
if (streamsConfiguration != null && streamsConfiguration.containsKey("application.server")) {
String applicationServer = (String) streamsConfiguration.get("application.server");
String[] appServerComponents = StringUtils.split(applicationServer, ":");
if (appServerComponents != null) {
return new HostInfo(appServerComponents[0], Integer.parseInt(appServerComponents[1]));
}
return HostInfo.buildFromEndpoint(applicationServer);
}
return null;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -196,6 +196,17 @@ void currentHostInfoWithNoProperties() {
assertThat(currentKafkaStreamsApplicationHostInfo).isNull();
}

@Test
void currentHostInfoWithIpv6ApplicationServer() {
StreamsBuilderFactoryBean sbfb = new StreamsBuilderFactoryBean(new KafkaStreamsConfiguration(
Map.of(StreamsConfig.APPLICATION_ID_CONFIG, "iqs-ipv6",
StreamsConfig.APPLICATION_SERVER_CONFIG, "[::1]:8080")));
sbfb.setAutoStartup(false);
KafkaStreamsInteractiveQueryService iqs = new KafkaStreamsInteractiveQueryService(sbfb);
HostInfo currentKafkaStreamsApplicationHostInfo = iqs.getCurrentKafkaStreamsApplicationHostInfo();
assertThat(currentKafkaStreamsApplicationHostInfo).isEqualTo(new HostInfo("::1", 8080));
}

@Test
void hostInfoForKeyAndStore() throws Exception {
ensureKafkaStreamsProcessorIsUpAndRunning();
Expand Down
Loading