Skip to content

Commit

Permalink
[Transform] fix NPE in derive stats if shouldStopAtNextCheckpo… (#52940)
Browse files Browse the repository at this point in the history 
fixes a NPE in _stats in case shouldStopAtNextCheckpoint is set.
  • Loading branch information
Hendrik Muhs committed Mar 2, 2020
1 parent d102158 commit 49f41d1
Show file tree
Hide file tree
Showing 2 changed files with 109 additions and 50 deletions.
3 changes: 2 additions & 1 deletion .../main/java/org/elasticsearch/xpack/transform/action/TransportGetTransformStatsAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
Expand Down Expand Up @@ -189,7 +190,7 @@ static TransformStats deriveStats(TransformTask task, @Nullable TransformCheckpo
&& derivedState.equals(TransformStats.State.STOPPED) == false
&& derivedState.equals(TransformStats.State.FAILED) == false) {
derivedState = TransformStats.State.STOPPING;
reason = reason.isEmpty() ? "transform is set to stop at the next checkpoint" : reason;
reason = Strings.isNullOrEmpty(reason) ? "transform is set to stop at the next checkpoint" : reason;
}
return new TransformStats(
task.getTransformId(),
Expand Down
156 changes: 107 additions & 49 deletions .../java/org/elasticsearch/xpack/transform/action/TransportGetTransformStatsActionTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,101 +28,159 @@ public class TransportGetTransformStatsActionTests extends ESTestCase {

public void testDeriveStatsStopped() {
String transformId = "transform-with-stats";
String reason = "";
String reason = null;
TransformIndexerStats stats = TransformIndexerStatsTests.randomStats();
TransformState stoppedState =
new TransformState(TransformTaskState.STOPPED, IndexerState.STOPPED, null, 0, reason, null, null, true);
TransformState stoppedState = new TransformState(
TransformTaskState.STOPPED,
IndexerState.STOPPED,
null,
0,
reason,
null,
null,
true
);
withIdStateAndStats(transformId, stoppedState, stats);
TransformCheckpointingInfo info = new TransformCheckpointingInfo(
new TransformCheckpointStats(1, null, null, 1, 1),
new TransformCheckpointStats(2, null, null, 2, 5),
2,
Instant.now());

assertThat(TransportGetTransformStatsAction.deriveStats(task, null),
equalTo(new TransformStats(transformId, TransformStats.State.STOPPED, "", null, stats, TransformCheckpointingInfo.EMPTY)));
assertThat(TransportGetTransformStatsAction.deriveStats(task, info),
equalTo(new TransformStats(transformId, TransformStats.State.STOPPED, "", null, stats, info)));

Instant.now()
);

assertThat(
TransportGetTransformStatsAction.deriveStats(task, null),
equalTo(new TransformStats(transformId, TransformStats.State.STOPPED, reason, null, stats, TransformCheckpointingInfo.EMPTY))
);
assertThat(
TransportGetTransformStatsAction.deriveStats(task, info),
equalTo(new TransformStats(transformId, TransformStats.State.STOPPED, reason, null, stats, info))
);

reason = "foo";
stoppedState = new TransformState(TransformTaskState.STOPPED, IndexerState.STOPPED, null, 0, reason, null, null, true);
withIdStateAndStats(transformId, stoppedState, stats);

assertThat(TransportGetTransformStatsAction.deriveStats(task, null),
equalTo(new TransformStats(transformId, TransformStats.State.STOPPED, reason, null, stats, TransformCheckpointingInfo.EMPTY)));
assertThat(TransportGetTransformStatsAction.deriveStats(task, info),
equalTo(new TransformStats(transformId, TransformStats.State.STOPPED, reason, null, stats, info)));
assertThat(
TransportGetTransformStatsAction.deriveStats(task, null),
equalTo(new TransformStats(transformId, TransformStats.State.STOPPED, reason, null, stats, TransformCheckpointingInfo.EMPTY))
);
assertThat(
TransportGetTransformStatsAction.deriveStats(task, info),
equalTo(new TransformStats(transformId, TransformStats.State.STOPPED, reason, null, stats, info))
);
}

public void testDeriveStatsFailed() {
String transformId = "transform-with-stats";
String reason = "";
String reason = null;
TransformIndexerStats stats = TransformIndexerStatsTests.randomStats();
TransformState failedState =
new TransformState(TransformTaskState.FAILED, IndexerState.STOPPED, null, 0, reason, null, null, true);
TransformState failedState = new TransformState(TransformTaskState.FAILED, IndexerState.STOPPED, null, 0, reason, null, null, true);
withIdStateAndStats(transformId, failedState, stats);
TransformCheckpointingInfo info = new TransformCheckpointingInfo(
new TransformCheckpointStats(1, null, null, 1, 1),
new TransformCheckpointStats(2, null, null, 2, 5),
2,
Instant.now());

assertThat(TransportGetTransformStatsAction.deriveStats(task, null),
equalTo(new TransformStats(transformId, TransformStats.State.FAILED, "", null, stats, TransformCheckpointingInfo.EMPTY)));
assertThat(TransportGetTransformStatsAction.deriveStats(task, info),
equalTo(new TransformStats(transformId, TransformStats.State.FAILED, "", null, stats, info)));

Instant.now()
);

assertThat(
TransportGetTransformStatsAction.deriveStats(task, null),
equalTo(new TransformStats(transformId, TransformStats.State.FAILED, reason, null, stats, TransformCheckpointingInfo.EMPTY))
);
assertThat(
TransportGetTransformStatsAction.deriveStats(task, info),
equalTo(new TransformStats(transformId, TransformStats.State.FAILED, reason, null, stats, info))
);

reason = "the task is failed";
failedState = new TransformState(TransformTaskState.FAILED, IndexerState.STOPPED, null, 0, reason, null, null, true);
withIdStateAndStats(transformId, failedState, stats);

assertThat(TransportGetTransformStatsAction.deriveStats(task, null),
equalTo(new TransformStats(transformId, TransformStats.State.FAILED, reason, null, stats, TransformCheckpointingInfo.EMPTY)));
assertThat(TransportGetTransformStatsAction.deriveStats(task, info),
equalTo(new TransformStats(transformId, TransformStats.State.FAILED, reason, null, stats, info)));
assertThat(
TransportGetTransformStatsAction.deriveStats(task, null),
equalTo(new TransformStats(transformId, TransformStats.State.FAILED, reason, null, stats, TransformCheckpointingInfo.EMPTY))
);
assertThat(
TransportGetTransformStatsAction.deriveStats(task, info),
equalTo(new TransformStats(transformId, TransformStats.State.FAILED, reason, null, stats, info))
);
}


public void testDeriveStats() {
String transformId = "transform-with-stats";
String reason = "";
String reason = null;
TransformIndexerStats stats = TransformIndexerStatsTests.randomStats();
TransformState runningState =
new TransformState(TransformTaskState.STARTED, IndexerState.INDEXING, null, 0, reason, null, null, true);
TransformState runningState = new TransformState(
TransformTaskState.STARTED,
IndexerState.INDEXING,
null,
0,
reason,
null,
null,
true
);
withIdStateAndStats(transformId, runningState, stats);
TransformCheckpointingInfo info = new TransformCheckpointingInfo(
new TransformCheckpointStats(1, null, null, 1, 1),
new TransformCheckpointStats(2, null, null, 2, 5),
2,
Instant.now());

assertThat(TransportGetTransformStatsAction.deriveStats(task, null),
equalTo(new TransformStats(transformId, TransformStats.State.STOPPING,
"transform is set to stop at the next checkpoint", null, stats, TransformCheckpointingInfo.EMPTY)));
assertThat(TransportGetTransformStatsAction.deriveStats(task, info),
equalTo(new TransformStats(transformId, TransformStats.State.STOPPING,
"transform is set to stop at the next checkpoint", null, stats, info)));

Instant.now()
);

assertThat(
TransportGetTransformStatsAction.deriveStats(task, null),
equalTo(
new TransformStats(
transformId,
TransformStats.State.STOPPING,
"transform is set to stop at the next checkpoint",
null,
stats,
TransformCheckpointingInfo.EMPTY
)
)
);
assertThat(
TransportGetTransformStatsAction.deriveStats(task, info),
equalTo(
new TransformStats(
transformId,
TransformStats.State.STOPPING,
"transform is set to stop at the next checkpoint",
null,
stats,
info
)
)
);

reason = "foo";
runningState = new TransformState(TransformTaskState.STARTED, IndexerState.INDEXING, null, 0, reason, null, null, true);
withIdStateAndStats(transformId, runningState, stats);

assertThat(TransportGetTransformStatsAction.deriveStats(task, null),
equalTo(new TransformStats(transformId, TransformStats.State.STOPPING, reason, null, stats, TransformCheckpointingInfo.EMPTY)));
assertThat(TransportGetTransformStatsAction.deriveStats(task, info),
equalTo(new TransformStats(transformId, TransformStats.State.STOPPING, reason, null, stats, info)));
assertThat(
TransportGetTransformStatsAction.deriveStats(task, null),
equalTo(new TransformStats(transformId, TransformStats.State.STOPPING, reason, null, stats, TransformCheckpointingInfo.EMPTY))
);
assertThat(
TransportGetTransformStatsAction.deriveStats(task, info),
equalTo(new TransformStats(transformId, TransformStats.State.STOPPING, reason, null, stats, info))
);

// Stop at next checkpoint is false.
runningState = new TransformState(TransformTaskState.STARTED, IndexerState.INDEXING, null, 0, reason, null, null, false);
withIdStateAndStats(transformId, runningState, stats);

assertThat(TransportGetTransformStatsAction.deriveStats(task, null),
equalTo(new TransformStats(transformId, TransformStats.State.INDEXING, reason, null, stats, TransformCheckpointingInfo.EMPTY)));
assertThat(TransportGetTransformStatsAction.deriveStats(task, info),
equalTo(new TransformStats(transformId, TransformStats.State.INDEXING, reason, null, stats, info)));
assertThat(
TransportGetTransformStatsAction.deriveStats(task, null),
equalTo(new TransformStats(transformId, TransformStats.State.INDEXING, reason, null, stats, TransformCheckpointingInfo.EMPTY))
);
assertThat(
TransportGetTransformStatsAction.deriveStats(task, info),
equalTo(new TransformStats(transformId, TransformStats.State.INDEXING, reason, null, stats, info))
);
}

private void withIdStateAndStats(String transformId, TransformState state, TransformIndexerStats stats) {
Expand Down

0 comments on commit 49f41d1

Please sign in to comment.