Skip to content

Commit 73db87e

Browse files
leehautleehaut
authored
Fix#7303 REST client silent failure when error hop is disabled (#7323)
* Fix REST client silent failure when error hop is disabled Signed-off-by: lance <leehaut@gmail.com> * Fix pipeline meta doc Signed-off-by: lance <leehaut@gmail.com> * Fix BeamFlinkPipelineEngineTest.testFlinkPipelineEngine error Signed-off-by: lance <leehaut@gmail.com> * Fix targetTransform NPE Signed-off-by: lance <leehaut@gmail.com> --------- Signed-off-by: lance <leehaut@gmail.com>
1 parent b97fa1b commit 73db87e

17 files changed

Lines changed: 394 additions & 63 deletions

File tree

‎engine/src/main/java/org/apache/hop/pipeline/PipelineMeta.java‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1631,6 +1631,46 @@ public void lookupReferencesAfterLoading() {
16311631
// This is rarely used, for example in getTableFields.getTableFields()
16321632
transformMeta.setParentPipelineMeta(this);
16331633
}
1634+
syncTransformErrorHandlingWithHops();
1635+
}
1636+
1637+
/**
1638+
* Align {@link TransformErrorMeta} with the enabled state of the hop to the error target.
1639+
* Pipelines saved before error hops were flagged in hop metadata can have error handling marked
1640+
* enabled while the hop to the target transform is disabled.
1641+
*/
1642+
public void syncTransformErrorHandlingWithHops() {
1643+
for (TransformMeta transformMeta : transforms) {
1644+
TransformErrorMeta errorMeta = transformMeta.getTransformErrorMeta();
1645+
if (errorMeta == null || errorMeta.getTargetTransform() == null) {
1646+
continue;
1647+
}
1648+
PipelineHopMeta hop = findPipelineHop(transformMeta, errorMeta.getTargetTransform(), true);
1649+
if (hop != null) {
1650+
errorMeta.setEnabled(hop.isEnabled());
1651+
}
1652+
applyDefaultErrorHandlingFieldNamesIfUnset(errorMeta);
1653+
}
1654+
}
1655+
1656+
/**
1657+
* Legacy pipelines often enable error handling without naming the extra error columns, so
1658+
* rejected rows carry no description field for downstream transforms (e.g. Write to log).
1659+
*/
1660+
private static void applyDefaultErrorHandlingFieldNamesIfUnset(TransformErrorMeta errorMeta) {
1661+
if (!errorMeta.isEnabled()) {
1662+
return;
1663+
}
1664+
if (!Utils.isEmpty(errorMeta.getNrErrorsValueName())
1665+
|| !Utils.isEmpty(errorMeta.getErrorDescriptionsValueName())
1666+
|| !Utils.isEmpty(errorMeta.getErrorFieldsValueName())
1667+
|| !Utils.isEmpty(errorMeta.getErrorCodesValueName())) {
1668+
return;
1669+
}
1670+
// error nr, error code, error description
1671+
errorMeta.setNrErrorsValueName(TransformErrorMeta.FIELD_ERROR_ROW);
1672+
errorMeta.setErrorCodesValueName(TransformErrorMeta.FIELD_ERROR_CODE);
1673+
errorMeta.setErrorDescriptionsValueName(TransformErrorMeta.FIELD_ERROR_DESCRIPTION);
16341674
}
16351675

16361676
/**

‎engine/src/main/java/org/apache/hop/pipeline/transform/BaseTransform.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
import java.util.Hashtable;
3232
import java.util.List;
3333
import java.util.Map;
34+
import java.util.Objects;
3435
import java.util.concurrent.CopyOnWriteArrayList;
3536
import java.util.concurrent.TimeUnit;
3637
import java.util.concurrent.atomic.AtomicBoolean;
@@ -1520,6 +1521,19 @@ public void handlePutError(
15201521
}
15211522
}
15221523
incrementLinesRejected();
1524+
if (!Utils.isEmpty(errorDescriptions)) {
1525+
logError(errorDescriptions);
1526+
}
1527+
} else if (transformErrorMeta.isEnabled()) {
1528+
String name =
1529+
Objects.nonNull(transformErrorMeta.getTargetTransform())
1530+
? transformErrorMeta.getTargetTransform().getName()
1531+
: "targetTransform is null";
1532+
logError(
1533+
BaseMessages.getString(
1534+
PKG, "BaseTransform.Exception.ErrorHandlingRouteUnavailable", name));
1535+
setErrors(1);
1536+
stopAll();
15231537
}
15241538

15251539
verifyRejectionRates();

‎engine/src/main/java/org/apache/hop/pipeline/transform/TransformErrorMeta.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,12 @@ public class TransformErrorMeta extends ChangedFlag implements Cloneable {
4545
public static final String XML_SOURCE_TRANSFORM_TAG = "source_transform";
4646
public static final String XML_TARGET_TRANSFORM_TAG = "target_transform";
4747

48+
/** error nr, error code, error description */
49+
public static final String FIELD_ERROR_ROW = "error_row";
50+
51+
public static final String FIELD_ERROR_DESCRIPTION = "error_description";
52+
public static final String FIELD_ERROR_CODE = "error_code";
53+
4854
/** The source transform that can send the error rows */
4955
@HopMetadataProperty(key = "source_transform", storeWithName = true, lookupInList = "transforms")
5056
private TransformMeta sourceTransform;

‎engine/src/main/java/org/apache/hop/pipeline/transform/TransformMeta.java‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -616,10 +616,18 @@ public boolean supportsErrorHandling() {
616616
* @return true if error handling is defined and enabled
617617
*/
618618
public boolean isDoingErrorHandling() {
619-
return transform.supportsErrorHandling()
620-
&& transformErrorMeta != null
621-
&& transformErrorMeta.getTargetTransform() != null
622-
&& transformErrorMeta.isEnabled();
619+
if (!transform.supportsErrorHandling()
620+
|| transformErrorMeta == null
621+
|| transformErrorMeta.getTargetTransform() == null
622+
|| !transformErrorMeta.isEnabled()) {
623+
return false;
624+
}
625+
626+
if (parentPipelineMeta != null) {
627+
return parentPipelineMeta.findPipelineHop(this, transformErrorMeta.getTargetTransform())
628+
!= null;
629+
}
630+
return true;
623631
}
624632

625633
/**

‎engine/src/main/resources/org/apache/hop/pipeline/transform/messages/messages_en_US.properties‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ BaseTransform.Exception.SourceTransformToReadFromCantRunInMultipleCopies=The sou
4444
BaseTransform.Exception.SourceTransformToReadFromDoesntExist=The source transform to read from [{0}] couldn''t be found.
4545
BaseTransform.Exception.TargetTransformToWriteToCantRunInMultipleCopies=The target transform [{0}] to write to can''t be run in multiple ({1}) copies.
4646
BaseTransform.Exception.TargetTransformToWriteToDoesntExist=The target transform [{0}] to write to doesn''t exist.
47+
BaseTransform.Exception.ErrorHandlingRouteUnavailable=Error handling is enabled for target transform [{0}] but no enabled hop routes errors to it. Re-enable the error handling hop or disable error handling on this transform.
4748
BaseTransform.Log.FinishedDispatching=Finished dispatching
4849
BaseTransform.Log.FoundInputRowset=Found input rowset [{0}]
4950
BaseTransform.Log.FoundOutputRowset=Found output rowset [{0}]

‎engine/src/test/java/org/apache/hop/pipeline/PipelineMetaTest.java‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@
8080
import org.w3c.dom.Node;
8181
import org.w3c.dom.NodeList;
8282

83+
/** Unit test for {@link PipelineMeta} */
8384
class PipelineMetaTest {
8485
public static final String TRANSFORM_NAME = "Any transform name";
8586

@@ -690,6 +691,71 @@ void testErrorHandlingSerialization() throws Exception {
690691
assertNotNull(copy.findTransform("T1").getTransformErrorMeta());
691692
}
692693

694+
@Test
695+
void syncTransformErrorHandlingWithHopsAlignsErrorMetaWithHopEnabledState() {
696+
TransformMeta source = new TransformMeta("REST client", new FakeMeta());
697+
TransformMeta errorTarget = new TransformMeta("Write to log error", new FakeMeta());
698+
pipelineMeta.addTransform(source);
699+
pipelineMeta.addTransform(errorTarget);
700+
701+
PipelineHopMeta errorHop = new PipelineHopMeta(source, errorTarget);
702+
errorHop.setEnabled(false);
703+
pipelineMeta.addPipelineHop(errorHop);
704+
705+
TransformErrorMeta errorMeta = new TransformErrorMeta(source, errorTarget);
706+
errorMeta.setEnabled(true);
707+
source.setTransformErrorMeta(errorMeta);
708+
709+
pipelineMeta.lookupReferencesAfterLoading();
710+
711+
assertFalse(errorMeta.isEnabled());
712+
assertFalse(source.isDoingErrorHandling());
713+
}
714+
715+
@Test
716+
void lookupReferencesAfterLoadingKeepsErrorHandlingActiveWhenHopEnabled() {
717+
TransformMeta source = new TransformMeta("REST client", new FakeMeta());
718+
TransformMeta errorTarget = new TransformMeta("Write to log error", new FakeMeta());
719+
pipelineMeta.addTransform(source);
720+
pipelineMeta.addTransform(errorTarget);
721+
722+
PipelineHopMeta errorHop = new PipelineHopMeta(source, errorTarget);
723+
errorHop.setEnabled(true);
724+
pipelineMeta.addPipelineHop(errorHop);
725+
726+
TransformErrorMeta errorMeta = new TransformErrorMeta(source, errorTarget);
727+
errorMeta.setEnabled(true);
728+
source.setTransformErrorMeta(errorMeta);
729+
730+
pipelineMeta.lookupReferencesAfterLoading();
731+
732+
assertTrue(errorMeta.isEnabled());
733+
assertTrue(source.isDoingErrorHandling());
734+
}
735+
736+
@Test
737+
void syncTransformErrorHandlingAppliesDefaultErrorFieldNamesForLegacyPipelines() {
738+
TransformMeta source = new TransformMeta("REST client", new FakeMeta());
739+
TransformMeta errorTarget = new TransformMeta("Write to log error", new FakeMeta());
740+
pipelineMeta.addTransform(source);
741+
pipelineMeta.addTransform(errorTarget);
742+
743+
PipelineHopMeta errorHop = new PipelineHopMeta(source, errorTarget);
744+
errorHop.setEnabled(true);
745+
pipelineMeta.addPipelineHop(errorHop);
746+
747+
TransformErrorMeta errorMeta = new TransformErrorMeta(source, errorTarget);
748+
errorMeta.setEnabled(true);
749+
source.setTransformErrorMeta(errorMeta);
750+
751+
pipelineMeta.syncTransformErrorHandlingWithHops();
752+
753+
assertEquals(TransformErrorMeta.FIELD_ERROR_ROW, errorMeta.getNrErrorsValueName());
754+
assertEquals(
755+
TransformErrorMeta.FIELD_ERROR_DESCRIPTION, errorMeta.getErrorDescriptionsValueName());
756+
assertEquals(TransformErrorMeta.FIELD_ERROR_CODE, errorMeta.getErrorCodesValueName());
757+
}
758+
693759
@Test
694760
void testPartitioningSerialization() throws Exception {
695761
IHopMetadataSerializer<PartitionSchema> schemaSerializer =

‎engine/src/test/java/org/apache/hop/pipeline/transform/BaseTransformTest.java‎

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,11 +67,12 @@
6767
import org.mockito.junit.jupiter.MockitoExtension;
6868
import org.mockito.stubbing.Answer;
6969

70+
/** Unit test for {@link BaseTransform} */
7071
@ExtendWith(MockitoExtension.class)
7172
class BaseTransformTest {
7273
private TransformMockHelper<ITransformMeta, ITransformData> mockHelper;
7374

74-
@Mock IRowHandler rowHandler;
75+
@Mock private IRowHandler rowHandler;
7576

7677
@BeforeEach
7778
void setup() {
@@ -302,6 +303,32 @@ void putErrorWithRowHandler() throws HopException {
302303
.putError(iRowMeta, objects, 3L, "desc", "field1,field2", "errorCode");
303304
}
304305

306+
@Test
307+
void putErrorStopsPipelineWhenErrorRowSetUnavailable() throws HopException {
308+
TransformMeta targetMeta = mock(TransformMeta.class);
309+
when(targetMeta.getName()).thenReturn("Write to log error");
310+
TransformErrorMeta errorMeta = new TransformErrorMeta(mockHelper.transformMeta, targetMeta);
311+
errorMeta.setEnabled(true);
312+
when(mockHelper.transformMeta.getTransformErrorMeta()).thenReturn(errorMeta);
313+
314+
BaseTransform<ITransformMeta, ITransformData> base =
315+
spy(
316+
new BaseTransform<>(
317+
mockHelper.transformMeta,
318+
mockHelper.iTransformMeta,
319+
mockHelper.iTransformData,
320+
0,
321+
mockHelper.pipelineMeta,
322+
mockHelper.pipeline));
323+
324+
IRowMeta iRowMeta = new RowMeta();
325+
Object[] objects = new Object[] {"Bob", "Col"};
326+
base.putError(iRowMeta, objects, 1L, "PKIX path building failed", null, "Rest001");
327+
328+
verify(base).stopAll();
329+
assertTrue(base.getErrors() > 0);
330+
}
331+
305332
@Test
306333
void putGetFromPutToDefaultRowHandlerMethods() throws HopException {
307334
BaseTransform<ITransformMeta, ITransformData> baseTransform =

‎engine/src/test/java/org/apache/hop/pipeline/transform/TransformMetaErrorHandlingTest.java‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.hop.pipeline.transform;
19+
20+
import static org.junit.jupiter.api.Assertions.assertFalse;
21+
import static org.junit.jupiter.api.Assertions.assertTrue;
22+
23+
import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
24+
import org.apache.hop.pipeline.PipelineHopMeta;
25+
import org.apache.hop.pipeline.PipelineMeta;
26+
import org.apache.hop.pipeline.transform.transforms.FakeMeta;
27+
import org.junit.jupiter.api.Test;
28+
import org.junit.jupiter.api.extension.ExtendWith;
29+
30+
/** Tests for error-handling metadata vs hop enabled state (issue #7303). */
31+
@ExtendWith(RestoreHopEngineEnvironmentExtension.class)
32+
class TransformMetaErrorHandlingTest {
33+
34+
@Test
35+
void isDoingErrorHandlingFalseWhenErrorHopDisabled() {
36+
PipelineMeta pipelineMeta = new PipelineMeta();
37+
TransformMeta source = new TransformMeta("REST client", new FakeMeta());
38+
TransformMeta errorTarget = new TransformMeta("Write to log error", new FakeMeta());
39+
pipelineMeta.addTransform(source);
40+
pipelineMeta.addTransform(errorTarget);
41+
42+
PipelineHopMeta errorHop = new PipelineHopMeta(source, errorTarget);
43+
errorHop.setEnabled(false);
44+
pipelineMeta.addPipelineHop(errorHop);
45+
46+
TransformErrorMeta errorMeta = new TransformErrorMeta(source, errorTarget);
47+
errorMeta.setEnabled(true);
48+
source.setTransformErrorMeta(errorMeta);
49+
source.setParentPipelineMeta(pipelineMeta);
50+
51+
assertFalse(source.isDoingErrorHandling());
52+
}
53+
54+
@Test
55+
void isDoingErrorHandlingTrueWhenErrorHopEnabled() {
56+
PipelineMeta pipelineMeta = new PipelineMeta();
57+
TransformMeta source = new TransformMeta("REST client", new FakeMeta());
58+
TransformMeta errorTarget = new TransformMeta("Write to log error", new FakeMeta());
59+
pipelineMeta.addTransform(source);
60+
pipelineMeta.addTransform(errorTarget);
61+
62+
PipelineHopMeta errorHop = new PipelineHopMeta(source, errorTarget);
63+
errorHop.setEnabled(true);
64+
pipelineMeta.addPipelineHop(errorHop);
65+
66+
TransformErrorMeta errorMeta = new TransformErrorMeta(source, errorTarget);
67+
errorMeta.setEnabled(true);
68+
source.setTransformErrorMeta(errorMeta);
69+
source.setParentPipelineMeta(pipelineMeta);
70+
71+
assertTrue(source.isDoingErrorHandling());
72+
}
73+
74+
@Test
75+
void isDoingErrorHandlingUsesErrorMetaWhenNoParentPipeline() {
76+
TransformMeta source = new TransformMeta("REST client", new FakeMeta());
77+
TransformMeta errorTarget = new TransformMeta("Write to log error", new FakeMeta());
78+
79+
TransformErrorMeta errorMeta = new TransformErrorMeta(source, errorTarget);
80+
errorMeta.setEnabled(true);
81+
source.setTransformErrorMeta(errorMeta);
82+
83+
assertTrue(source.isDoingErrorHandling());
84+
}
85+
}

0 commit comments

Comments
 (0)