Skip to content
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.

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

[HUDI-7763] Fix that multiple jmx reporter can exist if metadata enables #11226

Merged
merged 4 commits into from
Jun 27, 2024
Merged
Show file tree
Hide file tree
Changes from 2 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 @@ -22,6 +22,7 @@
import org.apache.hudi.common.testutils.NetworkTestUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.config.metrics.HoodieMetricsConfig;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.storage.StorageConfiguration;

import org.junit.jupiter.api.AfterEach;
Expand All @@ -34,6 +35,8 @@
import java.util.UUID;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.when;

/**
Expand Down Expand Up @@ -80,4 +83,55 @@ public void testRegisterGaugeByRangerPort() {
assertEquals("123", metrics.getRegistry().getGauges()
.get("jmx_metric2").getValue().toString());
}

@Test
public void testMultipleJmxReporterServer() {
String ports = "9889-9890";
clearInvocations(metricsConfig);
when(metricsConfig.getMetricsReporterType()).thenReturn(MetricsReporterType.JMX);
when(metricsConfig.getJmxHost()).thenReturn("localhost");
when(metricsConfig.getJmxPort()).thenReturn(ports);
when(metricsConfig.getBasePath()).thenReturn("s3://test" + UUID.randomUUID());
hoodieMetrics = new HoodieMetrics(writeConfig, storageConf);
metrics = hoodieMetrics.getMetrics();

clearInvocations(metricsConfig);
when(metricsConfig.getMetricsReporterType()).thenReturn(MetricsReporterType.JMX);
when(metricsConfig.getJmxHost()).thenReturn("localhost");
when(metricsConfig.getJmxPort()).thenReturn(ports);
when(metricsConfig.getBasePath()).thenReturn("s3://test2" + UUID.randomUUID());

hoodieMetrics = new HoodieMetrics(writeConfig, storageConf);
Metrics metrics2 = hoodieMetrics.getMetrics();

metrics.registerGauge("jmx_metric3", 123L);
assertEquals("123", metrics.getRegistry().getGauges()
.get("jmx_metric3").getValue().toString());

metrics2.registerGauge("jmx_metric4", 123L);
assertEquals("123", metrics2.getRegistry().getGauges()
.get("jmx_metric4").getValue().toString());
}

@Test
public void testMultipleJmxReporterServerFailedForOnePort() {
String ports = "9891";
clearInvocations(metricsConfig);
when(metricsConfig.getMetricsReporterType()).thenReturn(MetricsReporterType.JMX);
when(metricsConfig.getJmxHost()).thenReturn("localhost");
when(metricsConfig.getJmxPort()).thenReturn(ports);
when(metricsConfig.getBasePath()).thenReturn("s3://test" + UUID.randomUUID());
hoodieMetrics = new HoodieMetrics(writeConfig, storageConf);
metrics = hoodieMetrics.getMetrics();

clearInvocations(metricsConfig);
when(metricsConfig.getMetricsReporterType()).thenReturn(MetricsReporterType.JMX);
when(metricsConfig.getJmxHost()).thenReturn("localhost");
when(metricsConfig.getJmxPort()).thenReturn(ports);
when(metricsConfig.getBasePath()).thenReturn("s3://test2" + UUID.randomUUID());

assertThrows(HoodieException.class, () -> {
hoodieMetrics = new HoodieMetrics(writeConfig, storageConf);
});
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import javax.management.MBeanServer;

import java.lang.management.ManagementFactory;
import java.rmi.server.ExportException;
import java.util.Objects;
import java.util.stream.IntStream;

Expand All @@ -53,16 +54,13 @@ public JmxMetricsReporter(HoodieMetricsConfig config, MetricRegistry registry) {
host, portsConfig));
}
int[] ports = getPortRangeFromString(portsConfig);
boolean successfullyStartedServer = false;
for (int port : ports) {
jmxReporterServer = createJmxReport(host, port);
LOG.info("Started JMX server on port " + port + ".");
successfullyStartedServer = true;
break;
}
initializeJmxReporterServer(host, ports);

boolean successfullyStartedServer = isServerCreated();
if (!successfullyStartedServer) {
throw new HoodieException(
"Could not start JMX server on any configured port. Ports: " + portsConfig);
"Could not start JMX server on any configured port. Ports: " + portsConfig
+ ". Maybe require port range for multiple hoodie tables");
}
LOG.info("Configured JMXReporter with {port:" + portsConfig + "}");
} catch (Exception e) {
Expand All @@ -72,9 +70,29 @@ public JmxMetricsReporter(HoodieMetricsConfig config, MetricRegistry registry) {
}
}

private boolean isServerCreated() {
return jmxReporterServer != null;
}

private void initializeJmxReporterServer(String host, int[] ports) {
for (int port : ports) {
try {
jmxReporterServer = createJmxReport(host, port);
LOG.info("Started JMX server on port " + port + ".");
break;
} catch (Exception e) {
Copy link
Contributor

Choose a reason for hiding this comment

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

Should we catch specific exception type here?

Copy link
Contributor Author

@hwani3142 hwani3142 May 16, 2024

Choose a reason for hiding this comment

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

Should we catch specific exception type here?

c7b4028
Callee function catches all exceptions, then rethrow to HoodieException.
Thus, check causes and log messages in detail.

Thanks for comment

if (e.getCause() instanceof ExportException) {
LOG.info("Skip for initializing jmx port " + port + " because of already in use");
} else {
LOG.info("Failed to initialize jmx port " + port + ". " + e.getMessage());
}
Copy link
Contributor

Choose a reason for hiding this comment

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

LOG.info("Failed to initialize jmx port " + port + ". ", e); ?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

emit log if all unknown error occur

}
}
}

@Override
public void start() {
if (jmxReporterServer != null) {
if (isServerCreated()) {
jmxReporterServer.start();
} else {
LOG.error("Cannot start as the jmxReporter is null.");
Expand Down
Loading