gianm closed pull request #5961: [Backport] revert lambda conversion to fix
occasional jvm error
URL: https://github.com/apache/incubator-druid/pull/5961
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/server/src/main/java/io/druid/server/coordinator/DruidCoordinator.java
b/server/src/main/java/io/druid/server/coordinator/DruidCoordinator.java
index b86fc81fe12..4e5bdaa489a 100644
--- a/server/src/main/java/io/druid/server/coordinator/DruidCoordinator.java
+++ b/server/src/main/java/io/druid/server/coordinator/DruidCoordinator.java
@@ -672,58 +672,63 @@ public CoordinatorHistoricalManagerRunnable(final int
startingLeaderCounter)
super(
ImmutableList.of(
new DruidCoordinatorSegmentInfoLoader(DruidCoordinator.this),
- params -> {
- // Display info about all historical servers
- Iterable<ImmutableDruidServer> servers = FunctionalIterable
- .create(serverInventoryView.getInventory())
- .filter(DruidServer::segmentReplicatable)
- .transform(DruidServer::toImmutableDruidServer);
-
- if (log.isDebugEnabled()) {
- log.debug("Servers");
- for (ImmutableDruidServer druidServer : servers) {
- log.debug(" %s", druidServer);
- log.debug(" -- DataSources");
- for (ImmutableDruidDataSource druidDataSource :
druidServer.getDataSources()) {
- log.debug(" %s", druidDataSource);
+ new DruidCoordinatorHelper()
+ {
+ @Override
+ public DruidCoordinatorRuntimeParams
run(DruidCoordinatorRuntimeParams params)
+ {
+ // Display info about all historical servers
+ Iterable<ImmutableDruidServer> servers = FunctionalIterable
+ .create(serverInventoryView.getInventory())
+ .filter(DruidServer::segmentReplicatable)
+ .transform(DruidServer::toImmutableDruidServer);
+
+ if (log.isDebugEnabled()) {
+ log.debug("Servers");
+ for (ImmutableDruidServer druidServer : servers) {
+ log.debug(" %s", druidServer);
+ log.debug(" -- DataSources");
+ for (ImmutableDruidDataSource druidDataSource :
druidServer.getDataSources()) {
+ log.debug(" %s", druidDataSource);
+ }
}
}
- }
- // Find all historical servers, group them by subType and sort
by ascending usage
- final DruidCluster cluster = new DruidCluster();
- for (ImmutableDruidServer server : servers) {
- if (!loadManagementPeons.containsKey(server.getName())) {
- LoadQueuePeon loadQueuePeon =
taskMaster.giveMePeon(server);
- loadQueuePeon.start();
- log.info("Created LoadQueuePeon for server[%s].",
server.getName());
+ // Find all historical servers, group them by subType and
sort by ascending usage
+ final DruidCluster cluster = new DruidCluster();
+ for (ImmutableDruidServer server : servers) {
+ if (!loadManagementPeons.containsKey(server.getName())) {
+ LoadQueuePeon loadQueuePeon =
taskMaster.giveMePeon(server);
+ loadQueuePeon.start();
+ log.info("Created LoadQueuePeon for server[%s].",
server.getName());
- loadManagementPeons.put(server.getName(), loadQueuePeon);
+ loadManagementPeons.put(server.getName(), loadQueuePeon);
+ }
+
+ cluster.add(new ServerHolder(server,
loadManagementPeons.get(server.getName())));
}
- cluster.add(new ServerHolder(server,
loadManagementPeons.get(server.getName())));
- }
+ segmentReplicantLookup =
SegmentReplicantLookup.make(cluster);
- segmentReplicantLookup = SegmentReplicantLookup.make(cluster);
+ // Stop peons for servers that aren't there anymore.
+ final Set<String> disappeared =
Sets.newHashSet(loadManagementPeons.keySet());
+ for (ImmutableDruidServer server : servers) {
+ disappeared.remove(server.getName());
+ }
+ for (String name : disappeared) {
+ log.info("Removing listener for server[%s] which is no
longer there.", name);
+ LoadQueuePeon peon = loadManagementPeons.remove(name);
+ peon.stop();
+ }
- // Stop peons for servers that aren't there anymore.
- final Set<String> disappeared =
Sets.newHashSet(loadManagementPeons.keySet());
- for (ImmutableDruidServer server : servers) {
- disappeared.remove(server.getName());
+ return params.buildFromExisting()
+ .withDruidCluster(cluster)
+ .withDatabaseRuleManager(metadataRuleManager)
+ .withLoadManagementPeons(loadManagementPeons)
+
.withSegmentReplicantLookup(segmentReplicantLookup)
+
.withBalancerReferenceTimestamp(DateTimes.nowUtc())
+ .build();
}
- for (String name : disappeared) {
- log.info("Removing listener for server[%s] which is no
longer there.", name);
- LoadQueuePeon peon = loadManagementPeons.remove(name);
- peon.stop();
- }
-
- return params.buildFromExisting()
- .withDruidCluster(cluster)
- .withDatabaseRuleManager(metadataRuleManager)
- .withLoadManagementPeons(loadManagementPeons)
-
.withSegmentReplicantLookup(segmentReplicantLookup)
-
.withBalancerReferenceTimestamp(DateTimes.nowUtc())
- .build();
},
new DruidCoordinatorRuleRunner(DruidCoordinator.this),
new DruidCoordinatorCleanupUnneeded(DruidCoordinator.this),
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscr...@druid.apache.org
For additional commands, e-mail: dev-h...@druid.apache.org
