Skip to content

KAFKA-6123: Give client MetricsReporter auto-generated client.id #5383

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cmccabe merged 8 commits into from
Oct 3, 2018
Merged

KAFKA-6123: Give client MetricsReporter auto-generated client.id #5383

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
2b9de3f
KAFKA-6123: Give client MetricsReporter auto-generated client.id
KevinLiLu Jul 18, 2018
5b9b06a
Check clientId using startsWith in unit tests
KevinLiLu Jul 20, 2018
3e354d4
Use existing MockMetricsReporter for testing
KevinLiLu Aug 1, 2018
a810d67
Give auto-generated client.id to all client types
KevinLiLu Aug 1, 2018
c0dc4ce
Fix checkstyle empty statement
KevinLiLu Aug 1, 2018
0c70c05
Add package-private clientId getters for testing
KevinLiLu Aug 7, 2018
01a8844
Make MockMetricsReporter.clientId non-static
KevinLiLu Aug 7, 2018
ee59e69
Remove unused import
KevinLiLu Aug 10, 2018
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
3 changes: 2 additions & 1 deletion clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -337,7 +337,8 @@ static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcesso
config.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG),
config.getLong(AdminClientConfig.METADATA_MAX_AGE_CONFIG));
List<MetricsReporter> reporters = config.getConfiguredInstances(AdminClientConfig.METRIC_REPORTER_CLASSES_CONFIG,
MetricsReporter.class);
MetricsReporter.class,
Collections.singletonMap(AdminClientConfig.CLIENT_ID_CONFIG, clientId));
Map<String, String> metricTags = Collections.singletonMap("client-id", clientId);
MetricConfig metricConfig = new MetricConfig().samples(config.getInt(AdminClientConfig.METRICS_NUM_SAMPLES_CONFIG))
.timeWindow(config.getLong(AdminClientConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS)
Expand Down
7 changes: 6 additions & 1 deletion clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -677,7 +677,8 @@ private KafkaConsumer(ConsumerConfig config,
.recordLevel(Sensor.RecordingLevel.forName(config.getString(ConsumerConfig.METRICS_RECORDING_LEVEL_CONFIG)))
.tags(metricsTags);
List<MetricsReporter> reporters = config.getConfiguredInstances(ConsumerConfig.METRIC_REPORTER_CLASSES_CONFIG,
MetricsReporter.class);
MetricsReporter.class,
Collections.singletonMap(ConsumerConfig.CLIENT_ID_CONFIG, clientId));
reporters.add(new JmxReporter(JMX_PREFIX));
this.metrics = new Metrics(metricConfig, reporters, time);
this.retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
Expand Down Expand Up @@ -2211,4 +2212,8 @@ private void throwIfNoAssignorsConfigured() {
ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG + " configuration property");
}

// Visible for testing
String getClientId() {

@harshach harshach Aug 23, 2018

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: please add protected keyword

@ijuma ijuma Aug 30, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

protected gives more visibility than what's there now (package visibility) so not sure why we'd want to do that?

return clientId;
}
}
8 changes: 7 additions & 1 deletion clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -352,7 +352,8 @@ public KafkaProducer(Properties properties, Serializer<K> keySerializer, Seriali
.recordLevel(Sensor.RecordingLevel.forName(config.getString(ProducerConfig.METRICS_RECORDING_LEVEL_CONFIG)))
.tags(metricTags);
List<MetricsReporter> reporters = config.getConfiguredInstances(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG,
MetricsReporter.class);
MetricsReporter.class,
Collections.singletonMap(ProducerConfig.CLIENT_ID_CONFIG, clientId));
reporters.add(new JmxReporter(JMX_PREFIX));
this.metrics = new Metrics(metricConfig, reporters, time);
ProducerMetrics metricsRegistry = new ProducerMetrics(this.metrics);
Expand Down Expand Up @@ -1202,6 +1203,11 @@ private void throwIfNoTransactionManager() {
"by setting the " + ProducerConfig.TRANSACTIONAL_ID_CONFIG + " configuration property");
}

// Visible for testing
String getClientId() {

@harshach harshach Aug 23, 2018

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: please add protected keyword

@ijuma ijuma Aug 30, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

protected gives more visibility than what's there now (package visibility) so not sure why we'd want to do that?

return clientId;
}

private static class ClusterAndWaitTime {
final Cluster cluster;
final long waitedOnMetadataMs;
Expand Down
14 changes: 14 additions & 0 deletions clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,20 @@ public class KafkaConsumerTest {
@Rule
public ExpectedException expectedException = ExpectedException.none();

@Test
public void testMetricsReporterAutoGeneratedClientId() {
Properties props = new Properties();
props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
props.setProperty(ConsumerConfig.METRIC_REPORTER_CLASSES_CONFIG, MockMetricsReporter.class.getName());
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(
props, new StringDeserializer(), new StringDeserializer());

MockMetricsReporter mockMetricsReporter = (MockMetricsReporter) consumer.metrics.reporters().get(0);

Assert.assertEquals(consumer.getClientId(), mockMetricsReporter.clientId);
consumer.close();
}

@Test
public void testConstructorClose() {
Properties props = new Properties();
Expand Down
14 changes: 14 additions & 0 deletions clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,20 @@
@PowerMockIgnore("javax.management.*")
public class KafkaProducerTest {

@Test
public void testMetricsReporterAutoGeneratedClientId() {
Properties props = new Properties();
props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
props.setProperty(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, MockMetricsReporter.class.getName());
KafkaProducer<String, String> producer = new KafkaProducer<>(
props, new StringSerializer(), new StringSerializer());

MockMetricsReporter mockMetricsReporter = (MockMetricsReporter) producer.metrics.reporters().get(0);

Assert.assertEquals(producer.getClientId(), mockMetricsReporter.clientId);
producer.close();
}

@Test
public void testConstructorWithSerializers() {
Properties producerProps = new Properties();
Expand Down
3 changes: 3 additions & 0 deletions clients/src/test/java/org/apache/kafka/test/MockMetricsReporter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
*/
package org.apache.kafka.test;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.metrics.KafkaMetric;
import org.apache.kafka.common.metrics.MetricsReporter;

Expand All @@ -26,6 +27,7 @@
public class MockMetricsReporter implements MetricsReporter {
public static final AtomicInteger INIT_COUNT = new AtomicInteger(0);
public static final AtomicInteger CLOSE_COUNT = new AtomicInteger(0);
public String clientId;

public MockMetricsReporter() {
}
Expand All @@ -48,5 +50,6 @@ public void close() {

@Override
public void configure(Map<String, ?> configs) {
clientId = (String) configs.get(CommonClientConfigs.CLIENT_ID_CONFIG);
}
}
4 changes: 3 additions & 1 deletion ...runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerGroupMember.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,9 @@ public WorkerGroupMember(DistributedConfig config,
MetricConfig metricConfig = new MetricConfig().samples(config.getInt(CommonClientConfigs.METRICS_NUM_SAMPLES_CONFIG))
.timeWindow(config.getLong(CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS)
.tags(metricsTags);
List<MetricsReporter> reporters = config.getConfiguredInstances(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG, MetricsReporter.class);
List<MetricsReporter> reporters = config.getConfiguredInstances(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG,
MetricsReporter.class,
Collections.singletonMap(CommonClientConfigs.CLIENT_ID_CONFIG, clientId));
reporters.add(new JmxReporter(JMX_PREFIX));
this.metrics = new Metrics(metricConfig, reporters, time);
this.retryBackoffMs = config.getLong(CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG);
Expand Down
3 changes: 2 additions & 1 deletion streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -659,7 +659,8 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
.recordLevel(Sensor.RecordingLevel.forName(config.getString(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG)))
.timeWindow(config.getLong(StreamsConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS);
final List<MetricsReporter> reporters = config.getConfiguredInstances(StreamsConfig.METRIC_REPORTER_CLASSES_CONFIG,
MetricsReporter.class);
MetricsReporter.class,
Collections.singletonMap(StreamsConfig.CLIENT_ID_CONFIG, clientId));
reporters.add(new JmxReporter(JMX_PREFIX));
metrics = new Metrics(metricConfig, reporters, time);

Expand Down