Repository navigation
Conversation
…o JavaAdminClient
…o JavaAdminClient
…o JavaAdminClient
| protected static final long DEFAULT_EXPONENTIAL_SLEEP_MAX_DELAY_MS = 10000; | ||
| protected static final int MAX_RETRIES_ON_EXCEPTION = 5; | ||
| protected static final int DEFAULT_REPL_FACTOR = 2; | ||
| private static final int KAFKA_ADMIN_OPS_TIMEOUT_MS = 50000; |
There was a problem hiding this comment.
- Do we need to make this to be configurable by the user.
- Why do we need to hardcode this to
50 seconds? Isn't that too high?
There was a problem hiding this comment.
I don't think user need to configure it. But I think you are right - it is too high. I will change it to 10 seconds (in case brokers are loaded), to match today's max delay.
| this.metadataConsumer = metadataConsumer; | ||
|
|
||
| Properties props = createAdminClientProperties(); | ||
| LOG.info("New admin client with props:" + props); |
There was a problem hiding this comment.
Why do we need to use two different adminClient?
There was a problem hiding this comment.
Because deleteMessages() is only provided by the old kafka.admin.AdminClient. We will get rid of it when we move to the new Kafka (1.0)
| @Override | ||
| public boolean createStream(StreamSpec streamSpec) { | ||
| LOG.info("Creating Kafka topic: {} on system: {}", streamSpec.getPhysicalName(), streamSpec.getSystemName()); | ||
| final String REPL_FACTOR = "replication.factor"; |
There was a problem hiding this comment.
nit: Would be better to move these static final constants to beginning of the class.
There was a problem hiding this comment.
Usually yes, but this is kind of unfortunate hack, and it should be used only in this method. So I'd prefer to keep it here.
| String repl = streamConfig.get(REPL_FACTOR); | ||
| LOG.warn("replication.factor (topic={}, repl={}) is invalid config for AdminClient.createTopics. Removing it. Using kSpec repl factor {}", | ||
| kSpec.getPhysicalName(), repl, kSpec.getReplicationFactor()); | ||
| streamConfig.remove(REPL_FACTOR); |
There was a problem hiding this comment.
We should preserve the replicationFactor value defined in the configuration.
There was a problem hiding this comment.
We should pass it in the spec.
You cannot pass it to the configs - it will fail config validation.
…ystemAdmin.java Co-Authored-By: sborya <boryas@apache.org>
…o JavaAdminClient
In this PR, I've refactored create/clear streams methods.