KieServerDataSetManager { protected void registerQueriesWithRetry(String serverTemplateId, String serverInstanceId, Set<QueryDefinition> queryDefinitions) throws Exception { long waitLimit = 5 * 60 * 1000; long elapsed = 0; try { QueryServicesClient queryClient = kieServerIntegration.getAdminServerClient(serverTemplateId, serverInstanceId).getServicesClient(QueryServicesClient.class); while (elapsed < waitLimit) { try { Iterator<QueryDefinition> definitionIt = queryDefinitions.iterator(); while (definitionIt.hasNext()) { QueryDefinition definition = definitionIt.next(); queryClient.replaceQuery(definition); LOGGER.info("Query definition {} (type {}) successfully registered on kie server '{}'", definition.getName(), definition.getTarget(), serverInstanceId); definitionIt.remove(); } event.fire(new KieServerDataSetRegistered(serverInstanceId, serverTemplateId)); return; } catch (KieServicesException | KieServerHttpRequestException e) { Thread.sleep(500); elapsed += 500; queryClient = kieServerIntegration.getAdminServerClientCheckEndpoints(serverTemplateId).getServicesClient(QueryServicesClient.class); LOGGER.debug("Cannot reach KIE Server, elapsed time while waiting '{}', max time '{}' error {}", elapsed, waitLimit, e.getMessage()); } } LOGGER.warn("Timeout while trying to register query definitions on '{}'", serverInstanceId); } catch (KieServicesException ex) { LOGGER.info("Not possible to register queries on server {} most likely due to BPM capability missing (details {})", serverInstanceId, ex.getMessage()); } } @Inject  KieServerDataSetManager(DataSetDefRegistry dataSetDefRegistry,
                                   KieServerIntegration kieServerIntegration,
                                   Event<KieServerDataSetRegistered> event,
                                   @Managed ExecutorService executorService,
                                   SpecManagementService specManagementService); void registerInKieServer(@Observes final ServerInstanceRegistered serverInstanceRegistered);  }
@Test public void testRegisterQueriesWithoutRetry() throws Exception { QueryDefinition query = QueryDefinition.builder().name("test").expression("expression").source("jbpm").target("CUSTOM").build(); Set<QueryDefinition> definitions = new HashSet<>(); definitions.add(query); kieServerDataSetManager.registerQueriesWithRetry("template", "instance", definitions); verify(kieServerIntegration, times(1)).getAdminServerClient(anyString(), anyString()); verify(queryClient, times(1)).replaceQuery(any()); assertEquals(1, receivedEvents.size()); }
@Test public void testRegisterQueriesOnServerWithoutBPMCapability() throws Exception { when(kieClient.getServicesClient(any())).thenThrow(KieServicesException.class); QueryDefinition query = QueryDefinition.builder().name("test").expression("expression").source("jbpm").target("CUSTOM").build(); Set<QueryDefinition> definitions = new HashSet<>(); definitions.add(query); kieServerDataSetManager.registerQueriesWithRetry("template", "instance", definitions); verify(kieServerIntegration, times(1)).getAdminServerClient(anyString(), anyString()); verify(queryClient, never()).replaceQuery(any()); assertEquals(0, receivedEvents.size()); }