import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
-import org.elasticsearch.client.indices.GetIndexRequest;
+import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
@Autowired
private ApplicationConfiguration config;
-
+
@Autowired
private DbService dbService;
private RestHighLevelClient client;
ActionListener<BulkResponse> listener;
-
-//ES Encrypted communication https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/_encrypted_communication.html#_encrypted_communication
-//Basic authentication https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/_basic_authentication.html
+ //ES Encrypted communication https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/_encrypted_communication.html#_encrypted_communication
+ //Basic authentication https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/_basic_authentication.html
@PostConstruct
private void init() {
Db elasticsearch = dbService.getElasticsearch();
String elasticsearchHost = elasticsearch.getHost();
-
+
// Initialize the Connection
client = new RestHighLevelClient(RestClient.builder(new HttpHost(elasticsearchHost, 9200, "http"), new HttpHost(elasticsearchHost, 9201, "http")));
public void cleanUp() throws IOException {
client.close();
}
-
+
public void ensureTableExist(String topic) throws IOException {
String topicLower = topic.toLowerCase();
-
- GetIndexRequest request = new GetIndexRequest(topicLower);
-
+
+ GetIndexRequest request = new GetIndexRequest(topicLower);
+
boolean exists = client.indices().exists(request, RequestOptions.DEFAULT);
- if(!exists){
+ if (!exists) {
//TODO submit mapping template
- CreateIndexRequest createIndexRequest = new CreateIndexRequest(topicLower);
- CreateIndexResponse createIndexResponse = client.indices().create(createIndexRequest, RequestOptions.DEFAULT);
+ CreateIndexRequest createIndexRequest = new CreateIndexRequest(topicLower);
+ CreateIndexResponse createIndexResponse = client.indices().create(createIndexRequest, RequestOptions.DEFAULT);
log.info("{} : created {}", createIndexResponse.index(), createIndexResponse.isAcknowledged());
}
}
-
+
//TTL is not supported in Elasticsearch 5.0 and later, what can we do? FIXME
public void saveJsons(TopicConfig topic, List<JSONObject> jsons) {
BulkRequest request = new BulkRequest();
for (JSONObject json : jsons) {
- if(topic.isCorrelateClearedMessage()) {
+ if (topic.isCorrelateClearedMessage()) {
boolean found = correlateClearedMessage(topic, json);
- if(found) {
+ if (found) {
continue;
}
}
-
+
String id = topic.getMessageId(json); //id can be null
request.add(new IndexRequest(topic.getName().toLowerCase(), config.getElasticsearchType(), id).source(json.toString(), XContentType.JSON));
}
log.debug("saving text to topic = {}, batch count = {} ", topic, jsons.size());
-
- if(config.isAsync()) {
- client.bulkAsync(request, RequestOptions.DEFAULT, listener);
- }else {
+
+ if (config.isAsync()) {
+ client.bulkAsync(request, RequestOptions.DEFAULT, listener);
+ } else {
try {
client.bulk(request, RequestOptions.DEFAULT);
- } catch (IOException e) {
- log.error( topic.getName() , e);
+ } catch (IOException e) {
+ log.error(topic.getName(), e);
}
}
}
* @param json
* @return boolean
*
- * Because of query by id, The search API cannot be used for query.
- * The search API can only query all data or based on the fields in the source.
- * So use the get API, three parameters: index, type, document id
+ * Because of query by id, The search API cannot be used for query. The
+ * search API can only query all data or based on the fields in the
+ * source. So use the get API, three parameters: index, type, document
+ * id
*/
private boolean correlateClearedMessage(TopicConfig topic, JSONObject json) {
boolean found = false;
try {
eName = json.query("/event/commonEventHeader/eventName").toString();
- if (StringUtils.isNotBlank(eName)) {
-
- if (eName.endsWith("Cleared")) {
-
- String name = eName.substring(0, eName.length() - 7);
- String reportingEntityName = json.query("/event/commonEventHeader/reportingEntityName").toString();
- String specificProblem = json.query("/event/faultFields/specificProblem").toString();
-
- String id = null;
- StringBuilder stringBuilder = new StringBuilder();
- stringBuilder = stringBuilder.append(name).append('^').append(reportingEntityName).append('^').append(specificProblem);
-
- id = stringBuilder.toString();//example: id = "aaaa^cccc^bbbbb"
- String index = topic.getName().toLowerCase();
-
- //get
- GetRequest getRequest = new GetRequest(index, config.getElasticsearchType(), id);
-
- GetResponse getResponse = null;
- try {
- getResponse = client.get(getRequest, RequestOptions.DEFAULT);
- if (getResponse != null) {
-
- if (getResponse.isExists()) {
- String sourceAsString = getResponse.getSourceAsString();
- JSONObject jsonObject = new JSONObject(sourceAsString);
- jsonObject.getJSONObject("event").getJSONObject("faultFields").put("vfStatus", "closed");
- String jsonString = jsonObject.toString();
-
- //update
- IndexRequest request = new IndexRequest(index, config.getElasticsearchType(), id);
- request.source(jsonString, XContentType.JSON);
- IndexResponse indexResponse = null;
- try {
- indexResponse = client.index(request, RequestOptions.DEFAULT);
- found = true;
- } catch (IOException e) {
- log.error("save failure");
- }
- } else {
- log.error("The getResponse was not exists" );
+ if (StringUtils.isNotBlank(eName) && eName.endsWith("Cleared")) {
+
+ String name = eName.substring(0, eName.length() - 7);
+ String reportingEntityName = json.query("/event/commonEventHeader/reportingEntityName").toString();
+ String specificProblem = json.query("/event/faultFields/specificProblem").toString();
+
+ String id = null;
+ StringBuilder stringBuilder = new StringBuilder();
+ stringBuilder = stringBuilder.append(name).append('^').append(reportingEntityName).append('^').append(specificProblem);
+
+ id = stringBuilder.toString();//example: id = "aaaa^cccc^bbbbb"
+ String index = topic.getName().toLowerCase();
+
+ //get
+ GetRequest getRequest = new GetRequest(index, config.getElasticsearchType(), id);
+
+ GetResponse getResponse = null;
+ try {
+ getResponse = client.get(getRequest, RequestOptions.DEFAULT);
+ if (getResponse != null) {
+
+ if (getResponse.isExists()) {
+ String sourceAsString = getResponse.getSourceAsString();
+ JSONObject jsonObject = new JSONObject(sourceAsString);
+ jsonObject.getJSONObject("event").getJSONObject("faultFields").put("vfStatus", "closed");
+ String jsonString = jsonObject.toString();
+
+ //update
+ IndexRequest request = new IndexRequest(index, config.getElasticsearchType(), id);
+ request.source(jsonString, XContentType.JSON);
+ IndexResponse indexResponse = null;
+ try {
+ indexResponse = client.index(request, RequestOptions.DEFAULT);
+ found = true;
+ } catch (IOException e) {
+ log.error("save failure");
}
-
} else {
- log.error("The document for this id was not found" );
+ log.error("The getResponse was not exists");
}
- } catch (ElasticsearchException e) {
- if (e.status() == RestStatus.NOT_FOUND) {
- log.error("The document for this id was not found" );
- }
- if (e.status() == RestStatus.CONFLICT) {
- log.error("Version conflict" );
- }
- log.error("Get document exception", e);
- }catch (IOException e) {
- log.error(topic.getName() , e);
+ } else {
+ log.error("The document for this id was not found");
}
- } else {
- log.info("The data is normal");
+ } catch (ElasticsearchException e) {
+ if (e.status() == RestStatus.NOT_FOUND) {
+ log.error("The document for this id was not found");
+ }
+ if (e.status() == RestStatus.CONFLICT) {
+ log.error("Version conflict");
+ }
+ log.error("Get document exception", e);
+ } catch (IOException e) {
+ log.error(topic.getName(), e);
}
- } else {
- log.debug("event id null");
}
} catch (Exception e) {
- log.error("error",e);
+ log.error("error", e);
}
return found;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
+import com.mongodb.bulk.BulkWriteError;
+import com.mongodb.MongoBulkWriteException;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoClientOptions.Builder;
import com.mongodb.ServerAddress;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
+import com.mongodb.client.model.InsertManyOptions;
/**
* Service for using MongoDB
@Autowired
private ApplicationConfiguration config;
- private boolean dbReady = false;
+ private boolean dbReady = false;
@Autowired
private DbService dbService;
private MongoDatabase database;
private MongoClient mongoClient;
private Map<String, MongoCollection<Document>> mongoCollectionMap = new HashMap<>();
+ private InsertManyOptions insertManyOptions;
@PostConstruct
private void init() {
addrs.add(new ServerAddress(host, port)); // FIXME should be a list of address
-
try {
- if(StringUtils.isNoneBlank(userName) && StringUtils.isNoneBlank(password))
- {
+ if (StringUtils.isNoneBlank(userName) && StringUtils.isNoneBlank(password)) {
credential = MongoCredential.createCredential(userName, databaseName, password.toCharArray());
List<MongoCredential> credentialList = new ArrayList<MongoCredential>();
credentialList.add(credential);
mongoClient = new MongoClient(addrs, credentialList, options);
- }else
- {
+ } else {
mongoClient = new MongoClient(addrs, options);
}
- }catch(Exception ex){
+ } catch (Exception ex) {
dbReady = false;
log.error("Fail to initiate MongoDB" + mongodb.getHost());
return;
}
database = mongoClient.getDatabase(mongodb.getDatabase());
- dbReady = true;
+ insertManyOptions = new InsertManyOptions();
+ insertManyOptions.ordered(false);
+
+ dbReady = true;
}
@PreDestroy
}
public void saveJsons(TopicConfig topic, List<JSONObject> jsons) {
- if(dbReady == false)
+ if (dbReady == false)
return;
List<Document> documents = new ArrayList<>(jsons.size());
for (JSONObject json : jsons) {
String collectionName = topic.getName().replaceAll("[^a-zA-Z0-9]", "");//remove - _ .
MongoCollection<Document> collection = mongoCollectionMap.computeIfAbsent(collectionName, k -> database.getCollection(k));
- collection.insertMany(documents);
+
+ try {
+ collection.insertMany(documents, insertManyOptions);
+ } catch (MongoBulkWriteException e) {
+ List<BulkWriteError> bulkWriteErrors = e.getWriteErrors();
+ for (BulkWriteError bulkWriteError : bulkWriteErrors) {
+ log.error("Failed record: {}", bulkWriteError);
+ }
+ }
log.debug("saved text to topic = {}, batch count = {} ", topic, jsons.size());
}
Code Review / dcaegen2 / services.git / commitdiff