Skip to content

Commit bcc684b

Browse files
stroomdev66stroomdev66
authored
Merge pull request #4785 from gchq/gh-4784
#4784 Fix DynamicIndexingFilter values
2 parents b47e38c + 4cb184a commit bcc684b

File tree

14 files changed

+233
-237
lines changed

14 files changed

+233
-237
lines changed

Diff for: ‎stroom-annotation/stroom-annotation-pipeline/src/main/java/stroom/annotation/pipeline/AnnotationWriter.java

+23-40
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import stroom.pipeline.filter.AbstractXMLFilter;
2828
import stroom.pipeline.shared.data.PipelineElementType;
2929
import stroom.svg.shared.SvgImage;
30+
import stroom.util.CharBuffer;
3031
import stroom.util.shared.Severity;
3132

3233
import jakarta.inject.Inject;
@@ -90,20 +91,19 @@ class AnnotationWriter extends AbstractXMLFilter {
9091
private final LocationFactoryProxy locationFactory;
9192
private final ErrorReceiverProxy errorReceiverProxy;
9293
private final AnnotationCreator annotationCreator;
94+
private final CharBuffer content = new CharBuffer();
9395
private Locator locator;
9496

95-
private String lastTag = null;
96-
private Annotation currentAnnotation = null;
97+
private Annotation currentAnnotation = new Annotation();
9798
private String lastEventId = null;
9899
private String lastStreamId = null;
99100

100-
private ArrayList<EventId> currentEventIds = null;
101+
private ArrayList<EventId> currentEventIds = new ArrayList<>();
101102

102103
@Inject
103104
AnnotationWriter(final AnnotationCreator annotationCreator,
104105
final ErrorReceiverProxy errorReceiverProxy,
105-
final LocationFactoryProxy locationFactory
106-
) {
106+
final LocationFactoryProxy locationFactory) {
107107
this.errorReceiverProxy = errorReceiverProxy;
108108
this.locationFactory = locationFactory;
109109
this.annotationCreator = annotationCreator;
@@ -124,37 +124,12 @@ public void setDocumentLocator(final Locator locator) {
124124
@Override
125125
public void startElement(final String uri, final String localName, final String qName, final Attributes atts)
126126
throws SAXException {
127-
128-
lastTag = localName;
129-
130-
if (ANNOTATION_TAG.equals(localName)) {
131-
currentAnnotation = new Annotation();
132-
currentEventIds = new ArrayList<>();
133-
}
134-
127+
content.clear();
135128
super.startElement(uri, localName, qName, atts);
136129
}
137130

138-
@Override
139-
public void characters(char[] ch, int start, int length) throws SAXException {
140-
String val = new String(ch, start, length);
141-
142-
if (TITLE_TAG.equals(lastTag)) {
143-
currentAnnotation.setTitle(val);
144-
} else if (DESCRIPTION_TAG.equals(lastTag)) {
145-
currentAnnotation.setSubject(val);
146-
} else if (EVENTID_TAG.equals(lastTag)) {
147-
lastEventId = val;
148-
} else if (STREAMID_TAG.equals(lastTag)) {
149-
lastStreamId = val;
150-
}
151-
152-
super.characters(ch, start, length);
153-
}
154-
155131
@Override
156132
public void endElement(final String uri, final String localName, final String qName) throws SAXException {
157-
158133
if (EVENT_TAG.equals(localName)) {
159134
if (lastStreamId == null) {
160135
log(Severity.ERROR, "StreamId must be a long integer, but none provided ", null);
@@ -177,13 +152,7 @@ public void endElement(final String uri, final String localName, final String qN
177152
}
178153
}
179154
}
180-
} else if (ANNOTATION_TAG.equals(localName) && currentAnnotation != null) {
181-
CreateEntryRequest request = new CreateEntryRequest(
182-
currentAnnotation,
183-
Annotation.COMMENT,
184-
(StringEntryValue) null,
185-
currentEventIds);
186-
155+
} else if (ANNOTATION_TAG.equals(localName)) {
187156
try {
188157
annotationCreator.createEntry(new CreateEntryRequest(
189158
currentAnnotation,
@@ -194,13 +163,27 @@ public void endElement(final String uri, final String localName, final String qN
194163
} catch (final RuntimeException e) {
195164
log(Severity.ERROR, "Unable to create annotation " + currentAnnotation.getSubject(), e);
196165
}
197-
currentAnnotation = null;
198-
currentEventIds = null;
166+
currentAnnotation = new Annotation();
167+
currentEventIds = new ArrayList<>();
168+
} else if (TITLE_TAG.equals(localName)) {
169+
currentAnnotation.setTitle(content.toString());
170+
} else if (DESCRIPTION_TAG.equals(localName)) {
171+
currentAnnotation.setSubject(content.toString());
172+
} else if (EVENTID_TAG.equals(localName)) {
173+
lastEventId = content.toString();
174+
} else if (STREAMID_TAG.equals(localName)) {
175+
lastStreamId = content.toString();
199176
}
200177

178+
content.clear();
201179
super.endElement(uri, localName, qName);
202180
}
203181

182+
@Override
183+
public void characters(char[] ch, int start, int length) throws SAXException {
184+
content.append(ch, start, length);
185+
super.characters(ch, start, length);
186+
}
204187

205188
private void log(final Severity severity, final String message, final Exception e) {
206189
errorReceiverProxy.log(severity, locationFactory.create(locator), getElementId(), message, e);

Diff for: ‎stroom-kafka/stroom-kafka-impl/src/main/java/stroom/kafka/pipeline/StandardKafkaProducer.java

+45-39
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import stroom.pipeline.filter.AbstractXMLFilter;
3131
import stroom.pipeline.shared.data.PipelineElementType;
3232
import stroom.svg.shared.SvgImage;
33+
import stroom.util.CharBuffer;
3334
import stroom.util.logging.LambdaLogger;
3435
import stroom.util.logging.LambdaLoggerFactory;
3536
import stroom.util.shared.Severity;
@@ -51,6 +52,7 @@
5152
import java.time.format.DateTimeParseException;
5253
import java.util.ArrayDeque;
5354
import java.util.ArrayList;
55+
import java.util.Arrays;
5456
import java.util.List;
5557
import java.util.Queue;
5658
import java.util.concurrent.ExecutionException;
@@ -94,6 +96,7 @@ class StandardKafkaProducer extends AbstractXMLFilter {
9496
private final LocationFactoryProxy locationFactory;
9597
private final KafkaProducerFactory stroomKafkaProducerFactory;
9698
private final Queue<Future<RecordMetadata>> kafkaMetaFutures;
99+
private final CharBuffer content = new CharBuffer();
97100

98101
private Locator locator = null;
99102
private DocRef configRef = null;
@@ -143,7 +146,7 @@ public void startProcessing() {
143146

144147
kafkaProducer = sharedKafkaProducer.getKafkaProducer().orElseThrow(() -> {
145148
log(Severity.FATAL_ERROR, "No Kafka produce exists for config " + configRef, null);
146-
throw LoggedException.create("Unable to create Kafka Producer using config " + configRef);
149+
return LoggedException.create("Unable to create Kafka Producer using config " + configRef);
147150
});
148151
} catch (KafkaException ex) {
149152
log(Severity.FATAL_ERROR, "Unable to create Kafka Producer using config " + configRef.getUuid(), ex);
@@ -210,19 +213,19 @@ public void startElement(final String uri, final String localName, final String
210213
if (xmlValueDepth == -1) {
211214
final ErrorListener errorListener = new ErrorListener() {
212215
@Override
213-
public void warning(TransformerException exception) throws TransformerException {
216+
public void warning(TransformerException exception) {
214217
errorReceiverProxy.log(Severity.WARNING, locationFactory.create(locator), getElementId(),
215218
"Kafka XML value parse error", exception);
216219
}
217220

218221
@Override
219-
public void error(TransformerException exception) throws TransformerException {
222+
public void error(TransformerException exception) {
220223
errorReceiverProxy.log(Severity.ERROR, locationFactory.create(locator), getElementId(),
221224
"Kafka XML value parse error", exception);
222225
}
223226

224227
@Override
225-
public void fatalError(TransformerException exception) throws TransformerException {
228+
public void fatalError(TransformerException exception) {
226229
errorReceiverProxy.log(Severity.FATAL_ERROR, locationFactory.create(locator), getElementId(),
227230
"Kafka XML value parse error", exception);
228231
}
@@ -271,73 +274,76 @@ public void fatalError(TransformerException exception) throws TransformerExcepti
271274
}
272275
}
273276

274-
if (HEADER_ELEMENT_LOCAL_NAME.equals(localName)) {
275-
state.inHeader = true;
276-
}
277-
278277
if (state != null) {
278+
if (HEADER_ELEMENT_LOCAL_NAME.equals(localName)) {
279+
state.inHeader = true;
280+
}
279281
state.lastElement = localName;
280282
}
281283
}
282284

283-
285+
content.clear();
284286
super.startElement(uri, localName, qName, atts);
285287
}
286288

287289
@Override
288-
public void characters(char[] ch, int start, int length) throws SAXException {
289-
String val = new String(ch, start, length);
290-
String element = state.lastElement;
291-
if (KEY_ELEMENT_LOCAL_NAME.equals(element)) {
292-
if (state.inHeader) {
293-
state.headerNames.add(val);
294-
} else {
295-
state.key = val;
296-
}
297-
} else if (VALUE_ELEMENT_LOCAL_NAME.equals(element)) {
298-
if (state.inHeader) {
299-
state.headerVals.add(val);
300-
} else if (xmlValueDepth >= 0) {
301-
xmlValueHandler.characters(ch, start, length);
302-
} else {
303-
state.messageValue = val.getBytes(StandardCharsets.UTF_8);
290+
public void endElement(final String uri, final String localName, final String qName) throws SAXException {
291+
if (state != null) {
292+
if (KEY_ELEMENT_LOCAL_NAME.equals(localName)) {
293+
if (state.inHeader) {
294+
state.headerNames.add(content.toString());
295+
} else {
296+
state.key = content.toString();
297+
}
298+
} else if (VALUE_ELEMENT_LOCAL_NAME.equals(localName)) {
299+
if (state.inHeader) {
300+
state.headerVals.add(content.toString());
301+
} else if (xmlValueDepth < 0) {
302+
state.messageValue = content.toString().getBytes(StandardCharsets.UTF_8);
303+
}
304304
}
305305
}
306-
super.characters(ch, start, length);
307-
}
308-
309-
310-
@Override
311-
public void endElement(final String uri, final String localName, final String qName) throws SAXException {
312306

313307
if (xmlValueDepth == 0) {
314308
if (VALUE_ELEMENT_LOCAL_NAME.equals(localName)) {
315-
//Create the val from the contents of XML handler buffer
309+
// Create the val from the contents of XML handler buffer
316310
xmlValueHandler.endDocument();
317311
state.messageValue = outputStream.toByteArray();
318312
xmlValueDepth = -1;
319313
} else {
320314
throw new SAXException("Unexpected tag " + localName + " in kafka message value.");
321315
}
322316
} else if (xmlValueDepth > 0) {
323-
//Closing an XML value element
317+
// Closing an XML value element
324318
xmlValueHandler.endElement(uri, localName, qName);
325319
xmlValueDepth--;
326-
} else {
327-
if (state != null) {
328-
state.lastElement = null;
329-
}
330320

321+
} else if (state != null) {
322+
state.lastElement = null;
331323
if (HEADER_ELEMENT_LOCAL_NAME.equals(localName)) {
332324
state.inHeader = false;
333325
} else if (RECORD_ELEMENT_LOCAL_NAME.equals(localName)) {
334326
createKafkaMessage(state);
335327
state = null;
336328
}
337329
}
330+
331+
content.clear();
338332
super.endElement(uri, localName, qName);
339333
}
340334

335+
@Override
336+
public void characters(char[] ch, int start, int length) throws SAXException {
337+
if (state != null && VALUE_ELEMENT_LOCAL_NAME.equals(state.lastElement)) {
338+
if (!state.inHeader && xmlValueDepth >= 0) {
339+
xmlValueHandler.characters(ch, start, length);
340+
}
341+
}
342+
343+
content.append(ch, start, length);
344+
super.characters(ch, start, length);
345+
}
346+
341347
private void createKafkaMessage(KafkaMessageState state) {
342348

343349
if (state.isInvalid()) {
@@ -384,7 +390,7 @@ private void logState(final KafkaMessageState state) {
384390
}
385391
stringBuilder
386392
.append(" Value: ")
387-
.append(state.messageValue);
393+
.append(Arrays.toString(state.messageValue));
388394

389395
log(Severity.INFO, stringBuilder.toString(), null);
390396
}
@@ -400,7 +406,7 @@ public void setKafkaConfig(final DocRef configRef) {
400406
@SuppressWarnings("unused")
401407
@PipelineProperty(
402408
description = "At the end of the stream, wait for acknowledgement from the Kafka broker for all " +
403-
"the messages sent. This ensures errors are caught in the pipeline process.",
409+
"the messages sent. This ensures errors are caught in the pipeline process.",
404410
defaultValue = "true",
405411
displayPriority = 2)
406412
public void setFlushOnSend(final boolean flushOnSend) {

Diff for: ‎stroom-pipeline/src/main/java/stroom/pipeline/filter/ResultLocator.java

-1
Original file line numberDiff line numberDiff line change
@@ -142,7 +142,6 @@ public void endElement(final String uri, final String localName, final String qN
142142
@Override
143143
public void characters(final char[] ch, final int start, final int length) throws SAXException {
144144
colNo += length;
145-
146145
super.characters(ch, start, length);
147146
}
148147

Diff for: ‎stroom-pipeline/src/main/java/stroom/pipeline/filter/SchemaFilter.java

+7-8
Original file line numberDiff line numberDiff line change
@@ -354,7 +354,6 @@ public void endElement(final String uri, final String localName, final String qN
354354
public void characters(final char[] ch, final int start, final int length) throws SAXException {
355355
try {
356356
colNo += length;
357-
358357
if (validator != null) {
359358
validator.characters(ch, start, length);
360359
}
@@ -397,13 +396,13 @@ private boolean validateSchemaLocations(final String rootURI, final Map<String,
397396
final String validLocations = schemaSet.getLocations();
398397

399398
// Check that the root has a URI.
400-
if (rootURI == null || rootURI.trim().length() == 0) {
399+
if (rootURI == null || rootURI.trim().isEmpty()) {
401400
noNamespace(validLocations, schemaConstraint);
402401
return false;
403402
}
404403

405404
// Make sure we have some schema locations.
406-
if (schemaLocations == null || schemaLocations.size() == 0) {
405+
if (schemaLocations == null || schemaLocations.isEmpty()) {
407406
noSchemaLocations(validLocations, schemaConstraint);
408407
return false;
409408
}
@@ -484,7 +483,7 @@ private void printSchemaLocations(final CharBuffer sb, final String validLocatio
484483

485484
if (schemaConstraint != null) {
486485
if (!Strings.isNullOrEmpty(schemaConstraint.getSchemaGroup())) {
487-
if (where.length() == 0) {
486+
if (where.isEmpty()) {
488487
where.append(" where ");
489488
} else {
490489
where.append(" and ");
@@ -494,7 +493,7 @@ private void printSchemaLocations(final CharBuffer sb, final String validLocatio
494493
where.append("'");
495494
}
496495
if (!Strings.isNullOrEmpty(schemaConstraint.getNamespaceURI())) {
497-
if (where.length() == 0) {
496+
if (where.isEmpty()) {
498497
where.append(" where ");
499498
} else {
500499
where.append(" and ");
@@ -504,7 +503,7 @@ private void printSchemaLocations(final CharBuffer sb, final String validLocatio
504503
where.append("'");
505504
}
506505
if (!Strings.isNullOrEmpty(schemaConstraint.getSystemId())) {
507-
if (where.length() == 0) {
506+
if (where.isEmpty()) {
508507
where.append(" where ");
509508
} else {
510509
where.append(" and ");
@@ -559,7 +558,7 @@ private ValidatorHandler getValidator() throws SAXException {
559558
final String location = schemaLocations.get(namespace);
560559
if (location != null) {
561560
namespaces.append(" xmlns");
562-
if (prefix.length() > 0) {
561+
if (!prefix.isEmpty()) {
563562
namespaces.append(":");
564563
namespaces.append(prefix);
565564
}
@@ -676,7 +675,7 @@ private void unexpectedError(final String method, final Throwable t) {
676675
if (t.getMessage() != null) {
677676
message = t.getMessage();
678677
}
679-
if (message == null || message.trim().length() == 0) {
678+
if (message == null || message.trim().isEmpty()) {
680679
message = t.getClass().getName();
681680
}
682681
}

0 commit comments

Comments
 (0)