Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

MINOR: Fix bug encountered when restoring RUN SCRIPT command. (again) #2265

Merged
merged 2 commits into from
Dec 13, 2018
Merged

MINOR: Fix bug encountered when restoring RUN SCRIPT command. (again) #2265

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
458d8a9
MINOR: Fix bug encountered when restoring RUN SCRIPT command. (again)
vcrfxia Dec 13, 2018
a508889
add test for handleRunScript when in mode=EXECUTE.
vcrfxia Dec 13, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/StatementExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -222,7 +222,7 @@ private void executeStatement(
terminateQuery((TerminateQuery) statement, mode);
successMessage = "Query terminated.";
} else if (statement instanceof RunScript) {
handleRunScript(command);
handleRunScript(command, mode);
} else {
throw new Exception(String.format(
"Unexpected statement type: %s",
Expand All @@ -237,7 +237,7 @@ private void executeStatement(
putFinalStatus(commandId, queuedCommandStatus, successStatus);
}

private void handleRunScript(final Command command) {
private void handleRunScript(final Command command, final Mode mode) {

if (command.getOverwriteProperties().containsKey(KsqlConstants.RUN_SCRIPT_STATEMENTS_CONTENT)) {
final String queries =
Expand All @@ -251,10 +251,13 @@ private void handleRunScript(final Command command) {
ksqlConfig.overrideBreakingConfigsWithOriginalValues(command.getOriginalProperties()),
overriddenProperties
);
for (final QueryMetadata queryMetadata : queryMetadataList) {
if (queryMetadata instanceof PersistentQueryMetadata) {
final PersistentQueryMetadata persistentQueryMd = (PersistentQueryMetadata) queryMetadata;
persistentQueryMd.start();
if (mode == Mode.EXECUTE) {
for (final QueryMetadata queryMetadata : queryMetadataList) {
if (queryMetadata instanceof PersistentQueryMetadata) {
final PersistentQueryMetadata persistentQueryMd =
(PersistentQueryMetadata) queryMetadata;
persistentQueryMd.start();
}
}
}
} else {
Expand Down
64 changes: 64 additions & 0 deletions ...st-app/src/test/java/io/confluent/ksql/rest/server/computation/StatementExecutorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@
import io.confluent.ksql.parser.tree.QualifiedName;
import io.confluent.ksql.parser.tree.Query;
import io.confluent.ksql.parser.tree.QuerySpecification;
import io.confluent.ksql.parser.tree.RunScript;
import io.confluent.ksql.parser.tree.Statement;
import io.confluent.ksql.parser.tree.Table;
import io.confluent.ksql.parser.tree.TerminateQuery;
Expand All @@ -61,6 +62,7 @@
import io.confluent.ksql.schema.registry.MockSchemaRegistryClientFactory;
import io.confluent.ksql.util.FakeKafkaClientSupplier;
import io.confluent.ksql.util.KsqlConfig;
import io.confluent.ksql.util.KsqlConstants;
import io.confluent.ksql.util.Pair;
import io.confluent.ksql.util.PersistentQueryMetadata;
import java.util.Collections;
Expand Down Expand Up @@ -527,6 +529,18 @@ private PersistentQueryMetadata mockReplayCSAS(
return mockQuery;
}

private PersistentQueryMetadata mockReplayRunScript(
final String runScriptStatement, final String queryStatement) {
final PersistentQueryMetadata mockQuery = mock(PersistentQueryMetadata.class);
final Statement mockRunScript = mock(RunScript.class);
expect(mockParser.parseSingleStatement(runScriptStatement)).andReturn(mockRunScript);
expect(mockEngine.buildMultipleQueries(eq(queryStatement), anyObject(), anyObject()))
.andReturn(Collections.singletonList(mockQuery));
mockQuery.start();
expectLastCall().once();
return mockQuery;
}

@Test
public void shouldSkipStartForTerminatedQueriesWhenReplayingLog() {
// Given:
Expand Down Expand Up @@ -641,6 +655,56 @@ public void shouldNotCascadeDropStreamCommand() {
verify(mockParser, mockEngine, mockMetaStore);
}

@Test
public void shouldHandleRunScriptCommand() {
// Given:
final String runScriptStatement = "run script";
final String queryStatement = "a query";
final PersistentQueryMetadata mockQuery = mockReplayRunScript(runScriptStatement, queryStatement);
replayAll();

// When:
statementExecutorWithMocks.handleStatement(
new QueuedCommand(
new CommandId(CommandId.Type.STREAM, "RunScript", CommandId.Action.EXECUTE),
new Command(
runScriptStatement,
Collections
.singletonMap(KsqlConstants.RUN_SCRIPT_STATEMENTS_CONTENT, queryStatement),
Collections.emptyMap())
)
);

// Then:
verify(mockParser, mockEngine, mockQuery);
}

@Test
public void shouldRestoreRunScriptCommand() {
// Given:
final String runScriptStatement = "run script";
final String queryStatement = "a persistent query";
final PersistentQueryMetadata mockQuery = mockReplayRunScript(runScriptStatement, queryStatement);
expect(mockEngine.getPersistentQueries()).andReturn(Collections.singletonList(mockQuery));
replayAll();

// When:
statementExecutorWithMocks.handleRestoration(
Collections.singletonList(
new QueuedCommand(
new CommandId(CommandId.Type.STREAM, "RunScript", CommandId.Action.EXECUTE),
new Command(
runScriptStatement,
Collections.singletonMap(KsqlConstants.RUN_SCRIPT_STATEMENTS_CONTENT, queryStatement),
Collections.emptyMap())
)
)
);

// Then:
verify(mockParser, mockEngine, mockQuery);
}

private void createStreamsAndTables() {
final Command csCommand = new Command(
"CREATE STREAM pageview ("
Expand Down