lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3375185604
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -1056,6 +1055,7 @@ public Future<RecordMetadata> send(ProducerRecord<K, V>
record) {
* when send is invoked after producer has
been closed.
* @throws InterruptException If the thread is interrupted while
blocked
* @throws SerializationException If the key or value are not valid
objects given the configured serializers
+ * @throws BootstrapResolutionException if DNS resolution of the bootstrap
servers fails within {@code bootstrap.resolve.timeout.ms}
Review Comment:
the `sendOffsetsToTx` API is getting metadata-aware with KIP-1319 btw (this
PR https://github.com/apache/kafka/pull/22460). That means that it could start
throwing this BootstrapException once that PR and this KIP-909 PR go in I
expect, right?
We can update in either PR I expect, whichever goes in last so we have all
the parts (update just to mention the new exception in the java doc + update
the KIP). I'm reviewing both so it's covered, just sharing for awareness for
now. cc. @dajac
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java:
##########
@@ -426,6 +431,12 @@ public class ConsumerConfig extends AbstractConfig {
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY.toString()),
Importance.MEDIUM,
CommonClientConfigs.CLIENT_DNS_LOOKUP_DOC)
+ .define(BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG,
+ Type.LONG,
+
CommonClientConfigs.DEFAULT_BOOTSTRAP_RESOLVE_TIMEOUT_MS,
+ atLeast(0L),
+ Importance.MEDIUM,
Review Comment:
the KIP says HIGH for the config, what's the intention?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -157,13 +158,22 @@ public void poll(final long timeoutMs, final long
currentTimeMs) {
* @param onClose True when the network thread is closing.
*/
public void poll(final long timeoutMs, final long currentTimeMs, boolean
onClose) {
- trySend(currentTimeMs);
+ try {
+ trySend(currentTimeMs);
+ } catch (BootstrapResolutionException e) {
+ // Bootstrap DNS resolution timeout - propagate to app thread
+ backgroundEventHandler.add(new ErrorEvent(e));
Review Comment:
in the case where the boostrap timed out, and mainly thinking about the
async consumer, wouldn't we end up flooding the queue here adding ErrorEvents
in the background?
Wonder if we should propagate this via `metadata.fatalError` instead. The
async consumer was initially propagating the metadata errors with events like
this but changed to avoid this flooding issue and to ensure we could cancel
in-flight futures in the background before failing in the app thread API call
(IIRC)
This
https://github.com/apache/kafka/blob/5111ad253a7ac12061c73ce4ab62a19ca7dc3d4e/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java#L172
and maybeFailOnMetadataError on the ConsumerNetworkThread. Makes sense?
--update
for the ShareConsumer follow-up we need to think more about this, that one
uses events IIRC, so we need to think about the flooding there
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
