Update Kafka config dynamically #4316
Conversation
|
Although I did not expect anything to the contrary, I just wanted to add, that it also works with Docker for Windows on WSL backend. |
|
@kiview fixed 🎉 |
| @@ -135,7 +135,7 @@ | |||
| @NonNull | |||
| private String networkMode; | |||
|
|
|||
| @NonNull | |||
| @Nullable | |||
There was a problem hiding this comment.
the annotation was wrong (we do allow resetting the network to null (no network))
| // Kafka supports only one INTER_BROKER listener, so we have to pick one. | ||
| // The current algorithm uses the following order of resolving the IP: | ||
| // 1. Custom network's IP set via `withNetwork` | ||
| // 2. Bridge network's IP | ||
| // 3. Best effort fallback to getNetworkSettings#ipAddress | ||
| String ipAddress = containerInfo.getNetworkSettings().getNetworks().entrySet() | ||
| .stream() | ||
| .filter(it -> it.getValue().getIpAddress() != null) | ||
| .max(Comparator.comparingInt(entry -> { | ||
| if (getNetwork() != null && getNetwork().getId().equals(entry.getValue().getNetworkID())) { | ||
| return 2; | ||
| } | ||
|
|
||
| if ("bridge".equals(entry.getKey())) { | ||
| return 1; | ||
| } | ||
|
|
||
| return 0; | ||
| })) | ||
| .map(it -> it.getValue().getIpAddress()) | ||
| .orElseGet(() -> containerInfo.getNetworkSettings().getIpAddress()); | ||
|
|
||
| return String.format("BROKER://%s:%s", ipAddress, "9092"); | ||
| return String.format("BROKER://%s:%s", containerInfo.getConfig().getHostName(), "9092"); |
There was a problem hiding this comment.
this was super fun to discover, but apparently we over-engineered this, and hostName is enough to make it work reliably - it works with both "no network" (a.k.a. "bridge only") and multi network scenarios, as containers are visible by their hostname to other containers in the same network 😂
| command += ". /etc/confluent/docker/bash-config \n"; | ||
| command += "/etc/confluent/docker/configure \n"; | ||
| command += "/etc/confluent/docker/launch \n"; | ||
| // Optimization: skip the checks | ||
| command += "echo '' > /etc/confluent/docker/ensure \n"; | ||
| // Run the original command | ||
| command += "/etc/confluent/docker/run \n"; | ||
| withCommand("sh", "-c", command); |
There was a problem hiding this comment.
this change is a tiny bit unrelated but, after looking for ways of simplifying things, I figured that we could delegate to the original CMD of Confluent's Kafka image but optimize it by removing the ensure step that adds ~2s, as it needs to instantiate a ZooKeeper client (not really necessary in our case, we have our own checks)
| withEnv( | ||
| "KAFKA_ADVERTISED_LISTENERS", | ||
| String.format( | ||
| "BROKER://%s:9092", | ||
| getNetwork() != null | ||
| ? getNetworkAliases().get(0) | ||
| : "localhost" | ||
| ) | ||
| ); |
There was a problem hiding this comment.
if set to just "localhost", clustered Kafka would fail, because There Can Be Only One Localhost ™
|
Pinging @jstastny-cz / @gunnarmorling / @gmunozfe who reported #3932. While we haven't added the base impl that can be used (yet?), the new implementation is much easier and you may want to follow. Also, I tried it with Strimzi but then figured that, unlike with Confluent's Kafka, Strimzi does not seem to support |
|
FYI M1 still fails with: :( |
This PR dramatically simplifies our Kafka integration by updating Kafka's config on the fly, so that we don't need to generate the entry point script at all.
It may make it more compatible with M1 - need to test it separately.