Commit bbc99f13 authored by Jonas Waeber's avatar Jonas Waeber
Browse files

Improve query update report ids.

parent 06cf86ca
Pipeline #24212 passed with stages
in 4 minutes and 27 seconds
......@@ -51,6 +51,7 @@ class BulkProcessor(
private var recordsInBulk = 0
private val klaxon = Klaxon()
private val updateByQueryListener = UpdateByQueryListener(serviceName, producer)
private val listener = object : BulkProcessor.Listener {
var recordsCounter = 0
override fun beforeBulk(executionId: Long, request: BulkRequest) {
......@@ -111,7 +112,7 @@ class BulkProcessor(
log.error(message)
producer.sendReport(
Report(
executionId.toString(),
Service.getRandomReportId(),
ReportStatus.fatal,
m,
serviceName
......@@ -208,57 +209,6 @@ class BulkProcessor(
}
}
private val updateByQueryListener = object : ActionListener<BulkByScrollResponse> {
override fun onResponse(response: BulkByScrollResponse?) {
if (response != null) {
response.bulkFailures.forEach {
val report = Report(
it.id,
ReportStatus.fatal,
it.message,
serviceName
)
producer.sendReport(report, null)
}
producer.sendReport(
Report(
response.toString(),
ReportStatus.success,
"",
serviceName
), null
)
} else {
val report = Report(
"",
ReportStatus.fatal,
"No response from update request.",
serviceName
)
producer.sendReport(report, null)
}
}
override fun onFailure(e: java.lang.Exception?) {
val report = if (e == null) {
Report(
"",
ReportStatus.fatal,
"Failed to update documents.",
serviceName
)
} else {
Report(
"",
ReportStatus.fatal,
e.localizedMessage,
serviceName
)
}
producer.sendReport(report, null)
}
}
private fun addUpdateByQueryRequest(consumerRecord: ConsumerRecord<String, String>): Report {
val updateQuery = try {
klaxon.parse<UpdateQuery>(consumerRecord.value()).let {
......
......@@ -31,6 +31,13 @@ class Service(file: String = "app.yml") {
const val elasticIndex = "elastic.index"
const val elasticPort = "elastic.port"
const val elasticHost = "elastic.host"
fun getRandomReportId() : String {
val allowedChars = ('A'..'Z') + ('a'..'z') + ('0'..'9')
return (1..10)
.map { allowedChars.random() }
.joinToString("")
}
}
private val log = LogManager.getLogger("ElasticBulkActionService")
......
/*
* elasticsearch consumer service
* Copyright (C) 2020-2021 Jonas Waeber
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package org.memobase
import ch.memobase.reporting.Report
import ch.memobase.reporting.ReportStatus
import org.elasticsearch.action.ActionListener
import org.elasticsearch.index.reindex.BulkByScrollResponse
class UpdateByQueryListener(
private val serviceName: String,
private val producer: Producer
) : ActionListener<BulkByScrollResponse> {
override fun onResponse(response: BulkByScrollResponse?) {
if (response != null) {
response.bulkFailures.forEach {
val report = Report(
it.id,
ReportStatus.fatal,
it.message,
serviceName
)
producer.sendReport(report, null)
}
producer.sendReport(
Report(
Service.getRandomReportId(),
ReportStatus.success,
"",
serviceName
), null
)
} else {
val report = Report(
Service.getRandomReportId(),
ReportStatus.fatal,
"No response from update request.",
serviceName
)
producer.sendReport(report, null)
}
}
override fun onFailure(e: java.lang.Exception?) {
val report = if (e == null) {
Report(
Service.getRandomReportId(),
ReportStatus.fatal,
"Failed to update documents.",
serviceName
)
} else {
Report(
Service.getRandomReportId(),
ReportStatus.fatal,
e.localizedMessage,
serviceName
)
}
producer.sendReport(report, null)
}
}
\ No newline at end of file
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment