This is an automated email from the ASF dual-hosted git repository.
duanzhengqiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new aa68af01dcb Refactor optional named test cases (#17859)
aa68af01dcb is described below
commit aa68af01dcbfcf7743eacdd246820e9554ea2e55
Author: Liang Zhang <[email protected]>
AuthorDate: Sun May 22 20:49:55 2022 +0800
Refactor optional named test cases (#17859)
* Refactor PipelineJobExecutor
* Refactor ShardingSphereRuleMetaDataTest
* Refactor optional named test cases
* Refactor optional named test cases
---
...eDiscoveryRuleConfigurationYamlSwapperTest.java | 6 +-
...eSplittingRuleConfigurationYamlSwapperTest.java | 6 +-
.../shadow/route/engine/ShadowRouteEngine.java | 10 +--
...hardingRuleAlteredJobConfigurationPreparer.java | 4 +-
.../RowNumberPaginationContextEngineTest.java | 8 +--
.../rule/ShardingSphereRuleMetaDataTest.java | 10 +--
.../expression/impl/ListExpressionConverter.java | 8 +--
.../core/api/impl/RuleAlteredJobAPIImpl.java | 4 +-
.../pipeline/core/execute/PipelineJobExecutor.java | 77 ++++++++++------------
.../metadata/generator/PipelineDDLGenerator.java | 10 ++-
.../rulealtered/RuleAlteredJobPreparer.java | 6 +-
.../scenario/rulealtered/RuleAlteredJobWorker.java | 6 +-
.../service/SchemaMetaDataPersistServiceTest.java | 13 ++--
.../common/queryable/ShowAuthorityRuleHandler.java | 5 +-
.../common/queryable/ShowSQLParserRuleHandler.java | 20 +++---
.../admin/mysql/MySQLAdminExecutorFactoryTest.java | 23 +++----
.../impl/OpenGaussDCLStatementSQLVisitor.java | 8 +--
.../impl/PostgreSQLDCLStatementSQLVisitor.java | 8 +--
.../core/job/progress/JobProgressTest.java | 6 +-
19 files changed, 109 insertions(+), 129 deletions(-)
diff --git
a/shardingsphere-features/shardingsphere-db-discovery/shardingsphere-db-discovery-core/src/test/java/org/apache/shardingsphere/dbdiscovery/yaml/swapper/DatabaseDiscoveryRuleConfigurationYamlSwapperTest.java
b/shardingsphere-features/shardingsphere-db-discovery/shardingsphere-db-discovery-core/src/test/java/org/apache/shardingsphere/dbdiscovery/yaml/swapper/DatabaseDiscoveryRuleConfigurationYamlSwapperTest.java
index fb636efaabc..56944bc248d 100644
---
a/shardingsphere-features/shardingsphere-db-discovery/shardingsphere-db-discovery-core/src/test/java/org/apache/shardingsphere/dbdiscovery/yaml/swapper/DatabaseDiscoveryRuleConfigurationYamlSwapperTest.java
+++
b/shardingsphere-features/shardingsphere-db-discovery/shardingsphere-db-discovery-core/src/test/java/org/apache/shardingsphere/dbdiscovery/yaml/swapper/DatabaseDiscoveryRuleConfigurationYamlSwapperTest.java
@@ -84,11 +84,11 @@ public final class
DatabaseDiscoveryRuleConfigurationYamlSwapperTest {
}
private DatabaseDiscoveryRuleConfigurationYamlSwapper
getHARuleConfigurationYamlSwapper() {
- Optional<DatabaseDiscoveryRuleConfigurationYamlSwapper> optional =
YamlRuleConfigurationSwapperFactory.getAllInstances().stream()
+ Optional<DatabaseDiscoveryRuleConfigurationYamlSwapper> result =
YamlRuleConfigurationSwapperFactory.getAllInstances().stream()
.filter(each -> each instanceof
DatabaseDiscoveryRuleConfigurationYamlSwapper)
.map(each -> (DatabaseDiscoveryRuleConfigurationYamlSwapper)
each)
.findFirst();
- assertTrue(optional.isPresent());
- return optional.get();
+ assertTrue(result.isPresent());
+ return result.get();
}
}
diff --git
a/shardingsphere-features/shardingsphere-readwrite-splitting/shardingsphere-readwrite-splitting-core/src/test/java/org/apache/shardingsphere/readwritesplitting/swapper/ReadwriteSplittingRuleConfigurationYamlSwapperTest.java
b/shardingsphere-features/shardingsphere-readwrite-splitting/shardingsphere-readwrite-splitting-core/src/test/java/org/apache/shardingsphere/readwritesplitting/swapper/ReadwriteSplittingRuleConfigurationYamlSwapperTest.java
index 8c0c54c1ca8..525b92a7f33 100644
---
a/shardingsphere-features/shardingsphere-readwrite-splitting/shardingsphere-readwrite-splitting-core/src/test/java/org/apache/shardingsphere/readwritesplitting/swapper/ReadwriteSplittingRuleConfigurationYamlSwapperTest.java
+++
b/shardingsphere-features/shardingsphere-readwrite-splitting/shardingsphere-readwrite-splitting-core/src/test/java/org/apache/shardingsphere/readwritesplitting/swapper/ReadwriteSplittingRuleConfigurationYamlSwapperTest.java
@@ -98,11 +98,11 @@ public final class
ReadwriteSplittingRuleConfigurationYamlSwapperTest {
}
private ReadwriteSplittingRuleConfigurationYamlSwapper
getReadwriteSplittingRuleConfigurationYamlSwapper() {
- Optional<ReadwriteSplittingRuleConfigurationYamlSwapper> optional =
YamlRuleConfigurationSwapperFactory.getAllInstances().stream()
+ Optional<ReadwriteSplittingRuleConfigurationYamlSwapper> result =
YamlRuleConfigurationSwapperFactory.getAllInstances().stream()
.filter(each -> each instanceof
ReadwriteSplittingRuleConfigurationYamlSwapper)
.map(each -> (ReadwriteSplittingRuleConfigurationYamlSwapper)
each)
.findFirst();
- assertTrue(optional.isPresent());
- return optional.get();
+ assertTrue(result.isPresent());
+ return result.get();
}
}
diff --git
a/shardingsphere-features/shardingsphere-shadow/shardingsphere-shadow-core/src/main/java/org/apache/shardingsphere/shadow/route/engine/ShadowRouteEngine.java
b/shardingsphere-features/shardingsphere-shadow/shardingsphere-shadow-core/src/main/java/org/apache/shardingsphere/shadow/route/engine/ShadowRouteEngine.java
index 263c75c86ba..927b3b94c6d 100644
---
a/shardingsphere-features/shardingsphere-shadow/shardingsphere-shadow-core/src/main/java/org/apache/shardingsphere/shadow/route/engine/ShadowRouteEngine.java
+++
b/shardingsphere-features/shardingsphere-shadow/shardingsphere-shadow-core/src/main/java/org/apache/shardingsphere/shadow/route/engine/ShadowRouteEngine.java
@@ -46,12 +46,12 @@ public interface ShadowRouteEngine {
for (RouteUnit each : routeUnits) {
String logicName = each.getDataSourceMapper().getLogicName();
String shadowLogicName =
each.getDataSourceMapper().getActualName();
- Optional<String> sourceDataSourceNameOptional =
shadowRule.getSourceDataSourceName(shadowLogicName);
- if (sourceDataSourceNameOptional.isPresent()) {
- String sourceDataSourceName =
sourceDataSourceNameOptional.get();
- String shadowDataSourceName =
shadowDataSourceMappings.get(sourceDataSourceName);
+ Optional<String> sourceDataSourceName =
shadowRule.getSourceDataSourceName(shadowLogicName);
+ if (sourceDataSourceName.isPresent()) {
+ String shadowDataSourceName =
shadowDataSourceMappings.get(sourceDataSourceName.get());
toBeRemoved.add(each);
- toBeAdded.add(null == shadowDataSourceName ? new RouteUnit(new
RouteMapper(logicName, sourceDataSourceName), each.getTableMappers())
+ toBeAdded.add(null == shadowDataSourceName
+ ? new RouteUnit(new RouteMapper(logicName,
sourceDataSourceName.get()), each.getTableMappers())
: new RouteUnit(new RouteMapper(logicName,
shadowDataSourceName), each.getTableMappers()));
}
}
diff --git
a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/main/java/org/apache/shardingsphere/sharding/data/pipeline/ShardingRuleAlteredJobConfigurationPreparer.java
b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/main/java/org/apache/shardingsphere/sharding/data/pipeline/ShardingRuleAlteredJobConfigurationPreparer.java
index 12901478d4b..69f50967ed3 100644
---
a/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/main/java/org/apache/shardingsphere/sharding/data/pipeline/ShardingRuleAlteredJobConfigurationPreparer.java
+++
b/shardingsphere-features/shardingsphere-sharding/shardingsphere-sharding-core/src/main/java/org/apache/shardingsphere/sharding/data/pipeline/ShardingRuleAlteredJobConfigurationPreparer.java
@@ -150,9 +150,9 @@ public final class
ShardingRuleAlteredJobConfigurationPreparer implements RuleAl
ShardingSphereDatabase database =
PipelineContext.getContextManager().getMetaDataContexts().getDatabaseMetaData(jobConfig.getDatabaseName());
DumperConfiguration dumperConfig =
createDumperConfiguration(jobConfig.getDatabaseName(), dataSourceName,
dataSourcePropsMap.get(dataSourceName).getAllLocalProperties(), tableNameMap,
database);
- Optional<ShardingRuleConfiguration> targetRuleConfigOptional =
getTargetRuleConfiguration(jobConfig);
+ Optional<ShardingRuleConfiguration> targetRuleConfig =
getTargetRuleConfiguration(jobConfig);
Set<LogicTableName> reShardNeededTables =
jobConfig.splitLogicTableNames().stream().map(LogicTableName::new).collect(Collectors.toSet());
- Map<LogicTableName, Set<String>> shardingColumnsMap =
getShardingColumnsMap(targetRuleConfigOptional.orElse(sourceRuleConfig),
reShardNeededTables);
+ Map<LogicTableName, Set<String>> shardingColumnsMap =
getShardingColumnsMap(targetRuleConfig.orElse(sourceRuleConfig),
reShardNeededTables);
ImporterConfiguration importerConfig =
createImporterConfiguration(jobConfig, onRuleAlteredActionConfig,
shardingColumnsMap, database);
TaskConfiguration result = new TaskConfiguration(jobConfig,
dumperConfig, importerConfig);
log.info("createTaskConfiguration, dataSourceName={}, result={}",
dataSourceName, result);
diff --git
a/shardingsphere-infra/shardingsphere-infra-binder/src/test/java/org/apache/shardingsphere/infra/binder/segment/select/pagination/engine/RowNumberPaginationContextEngineTest.java
b/shardingsphere-infra/shardingsphere-infra-binder/src/test/java/org/apache/shardingsphere/infra/binder/segment/select/pagination/engine/RowNumberPaginationContextEngineTest.java
index 154e029893a..6d1a75602bf 100644
---
a/shardingsphere-infra/shardingsphere-infra-binder/src/test/java/org/apache/shardingsphere/infra/binder/segment/select/pagination/engine/RowNumberPaginationContextEngineTest.java
+++
b/shardingsphere-infra/shardingsphere-infra-binder/src/test/java/org/apache/shardingsphere/infra/binder/segment/select/pagination/engine/RowNumberPaginationContextEngineTest.java
@@ -120,11 +120,9 @@ public final class RowNumberPaginationContextEngineTest {
BinaryOperationExpression expression = new
BinaryOperationExpression(0, 0, left, right, operator, null);
PaginationContext paginationContext = new
RowNumberPaginationContextEngine().createPaginationContext(Collections.singletonList(expression),
projectionsContext, Collections.emptyList());
assertFalse(paginationContext.getOffsetSegment().isPresent());
- Optional<PaginationValueSegment> paginationValueSegmentOptional =
paginationContext.getRowCountSegment();
- assertTrue(paginationValueSegmentOptional.isPresent());
- PaginationValueSegment paginationValueSegment =
paginationValueSegmentOptional.get();
- assertThat(paginationValueSegment,
instanceOf(NumberLiteralRowNumberValueSegment.class));
- NumberLiteralRowNumberValueSegment numberLiteralRowNumberValueSegment
= (NumberLiteralRowNumberValueSegment) paginationValueSegment;
+ Optional<PaginationValueSegment> paginationValueSegment =
paginationContext.getRowCountSegment();
+ assertTrue(paginationValueSegment.isPresent());
+ NumberLiteralRowNumberValueSegment numberLiteralRowNumberValueSegment
= (NumberLiteralRowNumberValueSegment) paginationValueSegment.get();
assertThat(numberLiteralRowNumberValueSegment.getStartIndex(), is(0));
assertThat(numberLiteralRowNumberValueSegment.getStopIndex(), is(10));
assertThat(numberLiteralRowNumberValueSegment.getValue(), is(100L));
diff --git
a/shardingsphere-infra/shardingsphere-infra-common/src/test/java/org/apache/shardingsphere/infra/metadata/rule/ShardingSphereRuleMetaDataTest.java
b/shardingsphere-infra/shardingsphere-infra-common/src/test/java/org/apache/shardingsphere/infra/metadata/rule/ShardingSphereRuleMetaDataTest.java
index a36a85b3b0d..e4410b93879 100644
---
a/shardingsphere-infra/shardingsphere-infra-common/src/test/java/org/apache/shardingsphere/infra/metadata/rule/ShardingSphereRuleMetaDataTest.java
+++
b/shardingsphere-infra/shardingsphere-infra-common/src/test/java/org/apache/shardingsphere/infra/metadata/rule/ShardingSphereRuleMetaDataTest.java
@@ -27,15 +27,15 @@ import static org.junit.Assert.assertTrue;
public final class ShardingSphereRuleMetaDataTest {
- private final ShardingSphereRuleMetaData shardingSphereRuleMetaData = new
ShardingSphereRuleMetaData(Collections.emptyList(), Collections.singleton(new
ShardingSphereRuleFixture()));
+ private final ShardingSphereRuleMetaData ruleMetaData = new
ShardingSphereRuleMetaData(Collections.emptyList(), Collections.singleton(new
ShardingSphereRuleFixture()));
@Test
- public void assertFilterRulesReturnOneItem() {
-
assertThat(shardingSphereRuleMetaData.findRules(ShardingSphereRuleFixture.class).size(),
is(1));
+ public void assertFindRules() {
+
assertThat(ruleMetaData.findRules(ShardingSphereRuleFixture.class).size(),
is(1));
}
@Test
- public void assertFindSingleRuleReturnsIsPresent() {
-
assertTrue(shardingSphereRuleMetaData.findSingleRule(ShardingSphereRuleFixture.class).isPresent());
+ public void assertFindSingleRule() {
+
assertTrue(ruleMetaData.findSingleRule(ShardingSphereRuleFixture.class).isPresent());
}
}
diff --git
a/shardingsphere-infra/shardingsphere-infra-federation/shardingsphere-infra-federation-optimizer/src/main/java/org/apache/shardingsphere/infra/federation/optimizer/converter/segment/expression/impl/ListExpressionConverter.java
b/shardingsphere-infra/shardingsphere-infra-federation/shardingsphere-infra-federation-optimizer/src/main/java/org/apache/shardingsphere/infra/federation/optimizer/converter/segment/expression/impl/ListExpressionConverter.java
index a479c09fd83..4d9b2b94c59 100644
---
a/shardingsphere-infra/shardingsphere-infra-federation/shardingsphere-infra-federation-optimizer/src/main/java/org/apache/shardingsphere/infra/federation/optimizer/converter/segment/expression/impl/ListExpressionConverter.java
+++
b/shardingsphere-infra/shardingsphere-infra-federation/shardingsphere-infra-federation-optimizer/src/main/java/org/apache/shardingsphere/infra/federation/optimizer/converter/segment/expression/impl/ListExpressionConverter.java
@@ -39,15 +39,15 @@ public final class ListExpressionConverter implements
SQLSegmentConverter<ListEx
public Optional<SqlNode> convertToSQLNode(final ListExpression segment) {
SqlNode left = null;
for (ExpressionSegment each : segment.getItems()) {
- Optional<SqlNode> optional = new
ExpressionConverter().convertToSQLNode(each);
- if (!optional.isPresent()) {
+ Optional<SqlNode> sqlNode = new
ExpressionConverter().convertToSQLNode(each);
+ if (!sqlNode.isPresent()) {
continue;
}
if (null == left) {
- left = optional.get();
+ left = sqlNode.get();
continue;
}
- left = new SqlBasicCall(SqlStdOperatorTable.OR, new
SqlNode[]{left, optional.get()}, SqlParserPos.ZERO);
+ left = new SqlBasicCall(SqlStdOperatorTable.OR, new
SqlNode[]{left, sqlNode.get()}, SqlParserPos.ZERO);
}
return Optional.ofNullable(left);
}
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java
index a94242cfa3b..ac3cb12b716 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java
@@ -331,8 +331,8 @@ public final class RuleAlteredJobAPIImpl extends
AbstractPipelineJobAPIImpl impl
RuleAlteredContext ruleAlteredContext =
RuleAlteredJobWorker.createRuleAlteredContext(jobConfig);
GovernanceRepositoryAPI repositoryAPI =
PipelineAPIFactory.getGovernanceRepositoryAPI();
if (isDataConsistencyCheckNeeded(ruleAlteredContext)) {
- Optional<Boolean> checkResultOptional =
repositoryAPI.getJobCheckResult(jobId);
- if (!checkResultOptional.isPresent() ||
!checkResultOptional.get()) {
+ Optional<Boolean> checkResult =
repositoryAPI.getJobCheckResult(jobId);
+ if (!checkResult.isPresent() || !checkResult.get()) {
throw new PipelineVerifyFailedException("Data consistency
check is not finished or failed.");
}
}
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/PipelineJobExecutor.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/PipelineJobExecutor.java
index 9dd43a3c79a..72474b2834c 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/PipelineJobExecutor.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/PipelineJobExecutor.java
@@ -18,7 +18,6 @@
package org.apache.shardingsphere.data.pipeline.core.execute;
import lombok.extern.slf4j.Slf4j;
-import org.apache.shardingsphere.data.pipeline.api.RuleAlteredJobAPI;
import org.apache.shardingsphere.data.pipeline.api.RuleAlteredJobAPIFactory;
import
org.apache.shardingsphere.data.pipeline.api.config.rulealtered.RuleAlteredJobConfiguration;
import
org.apache.shardingsphere.data.pipeline.api.config.rulealtered.yaml.RuleAlteredJobConfigurationSwapper;
@@ -54,47 +53,7 @@ public final class PipelineJobExecutor extends
AbstractLifecycleExecutor {
@Override
protected void doStart() {
- watchGovernanceRepositoryConfiguration();
- }
-
- private void watchGovernanceRepositoryConfiguration() {
- RuleAlteredJobAPI ruleAlteredJobAPI =
RuleAlteredJobAPIFactory.getInstance();
-
PipelineAPIFactory.getGovernanceRepositoryAPI().watch(DataPipelineConstants.DATA_PIPELINE_ROOT,
event -> {
- Optional<JobConfigurationPOJO> jobConfigPOJOOptional =
getJobConfigPOJO(event);
- if (!jobConfigPOJOOptional.isPresent()) {
- return;
- }
- JobConfigurationPOJO jobConfigPOJO = jobConfigPOJOOptional.get();
- boolean deleted = DataChangedEvent.Type.DELETED == event.getType();
- boolean disabled = jobConfigPOJO.isDisabled();
- if (deleted || disabled) {
- log.info("jobId={}, deleted={}, disabled={}",
jobConfigPOJO.getJobName(), deleted, disabled);
- RuleAlteredJobSchedulerCenter.stop(jobConfigPOJO.getJobName());
- // TODO refactor: dispatch to different job types
- RuleAlteredJobConfiguration jobConfig =
RuleAlteredJobConfigurationSwapper.swapToObject(jobConfigPOJO.getJobParameter());
- if (deleted) {
- new RuleAlteredJobPreparer().cleanup(jobConfig);
- } else if
(RuleAlteredJobProgressDetector.isJobSuccessful(jobConfig.getJobShardingCount(),
ruleAlteredJobAPI.getProgress(jobConfig).values())) {
- log.info("isJobSuccessful=true");
- new RuleAlteredJobPreparer().cleanup(jobConfig);
- }
- ScalingReleaseDatabaseLevelLockEvent releaseLockEvent = new
ScalingReleaseDatabaseLevelLockEvent(jobConfig.getDatabaseName());
- ShardingSphereEventBus.getInstance().post(releaseLockEvent);
- return;
- }
- switch (event.getType()) {
- case ADDED:
- case UPDATED:
- if
(RuleAlteredJobSchedulerCenter.existJob(jobConfigPOJO.getJobName())) {
- log.info("{} added to executing jobs failed since it
already exists", jobConfigPOJO.getJobName());
- } else {
- executor.execute(() -> execute(jobConfigPOJO));
- }
- break;
- default:
- break;
- }
- });
+
PipelineAPIFactory.getGovernanceRepositoryAPI().watch(DataPipelineConstants.DATA_PIPELINE_ROOT,
event -> getJobConfigPOJO(event).ifPresent(optional -> processEvent(event,
optional)));
}
private Optional<JobConfigurationPOJO> getJobConfigPOJO(final
DataChangedEvent event) {
@@ -111,9 +70,39 @@ public final class PipelineJobExecutor extends
AbstractLifecycleExecutor {
return Optional.empty();
}
+ private void processEvent(final DataChangedEvent event, final
JobConfigurationPOJO jobConfigPOJO) {
+ boolean isDeleted = DataChangedEvent.Type.DELETED == event.getType();
+ boolean isDisabled = jobConfigPOJO.isDisabled();
+ if (isDeleted || isDisabled) {
+ log.info("jobId={}, deleted={}, disabled={}",
jobConfigPOJO.getJobName(), isDeleted, isDisabled);
+ RuleAlteredJobSchedulerCenter.stop(jobConfigPOJO.getJobName());
+ // TODO refactor: dispatch to different job types
+ RuleAlteredJobConfiguration jobConfig =
RuleAlteredJobConfigurationSwapper.swapToObject(jobConfigPOJO.getJobParameter());
+ if (isDeleted) {
+ new RuleAlteredJobPreparer().cleanup(jobConfig);
+ } else if
(RuleAlteredJobProgressDetector.isJobSuccessful(jobConfig.getJobShardingCount(),
RuleAlteredJobAPIFactory.getInstance().getProgress(jobConfig).values())) {
+ log.info("isJobSuccessful=true");
+ new RuleAlteredJobPreparer().cleanup(jobConfig);
+ }
+ ShardingSphereEventBus.getInstance().post(new
ScalingReleaseDatabaseLevelLockEvent(jobConfig.getDatabaseName()));
+ return;
+ }
+ switch (event.getType()) {
+ case ADDED:
+ case UPDATED:
+ if
(RuleAlteredJobSchedulerCenter.existJob(jobConfigPOJO.getJobName())) {
+ log.info("{} added to executing jobs failed since it
already exists", jobConfigPOJO.getJobName());
+ } else {
+ executor.execute(() -> execute(jobConfigPOJO));
+ }
+ break;
+ default:
+ break;
+ }
+ }
+
private void execute(final JobConfigurationPOJO jobConfigPOJO) {
- RuleAlteredJobConfiguration jobConfig =
RuleAlteredJobConfigurationSwapper.swapToObject(jobConfigPOJO.getJobParameter());
- String databaseName = jobConfig.getDatabaseName();
+ String databaseName =
RuleAlteredJobConfigurationSwapper.swapToObject(jobConfigPOJO.getJobParameter()).getDatabaseName();
if (PipelineSimpleLock.getInstance().tryLock(databaseName, 3000)) {
log.info("{} added to executing jobs success",
jobConfigPOJO.getJobName());
new OneOffJobBootstrap(PipelineAPIFactory.getRegistryCenter(), new
RuleAlteredJob(), jobConfigPOJO.toJobConfiguration()).execute();
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/generator/PipelineDDLGenerator.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/generator/PipelineDDLGenerator.java
index cb6426da1b2..62a6ed17457 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/generator/PipelineDDLGenerator.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/generator/PipelineDDLGenerator.java
@@ -127,13 +127,11 @@ public final class PipelineDDLGenerator {
private String generateActualDDLSQL(final DatabaseType databaseType, final
String schemaName, final String tableName, final ShardingSphereDatabase
database) throws SQLException {
DataNodes dataNodes = new
DataNodes(database.getRuleMetaData().getRules());
- Optional<DataNode> optional =
dataNodes.getDataNodes(tableName).stream()
- .filter(dataNode ->
database.getResource().getDataSources().containsKey(dataNode.getDataSourceName().contains(".")
- ? dataNode.getDataSourceName().split("\\.")[0]
- : dataNode.getDataSourceName()))
+ Optional<DataNode> filteredDataNode =
dataNodes.getDataNodes(tableName).stream()
+ .filter(each ->
database.getResource().getDataSources().containsKey(each.getDataSourceName().contains(".")
? each.getDataSourceName().split("\\.")[0] : each.getDataSourceName()))
.findFirst();
- String dataSourceName =
optional.map(DataNode::getDataSourceName).orElseGet(() ->
database.getResource().getDataSources().keySet().iterator().next());
- String actualTable =
optional.map(DataNode::getTableName).orElse(tableName);
+ String dataSourceName =
filteredDataNode.map(DataNode::getDataSourceName).orElseGet(() ->
database.getResource().getDataSources().keySet().iterator().next());
+ String actualTable =
filteredDataNode.map(DataNode::getTableName).orElse(tableName);
return
DialectDDLSQLGeneratorFactory.findInstance(databaseType).orElseThrow(() -> new
ShardingSphereException("Failed to get dialect ddl sql generator"))
.generateDDLSQL(actualTable, schemaName,
database.getResource().getDataSources().get(dataSourceName));
}
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobPreparer.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobPreparer.java
index a50b9318e0c..eaf7532443f 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobPreparer.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobPreparer.java
@@ -176,9 +176,9 @@ public final class RuleAlteredJobPreparer {
private IngestPosition<?> getIncrementalPosition(final
RuleAlteredJobContext jobContext, final TaskConfiguration taskConfig,
final
PipelineDataSourceManager dataSourceManager) throws SQLException {
if (null != jobContext.getInitProgress()) {
- Optional<IngestPosition<?>> positionOptional =
jobContext.getInitProgress().getIncrementalPosition(taskConfig.getDumperConfig().getDataSourceName());
- if (positionOptional.isPresent()) {
- return positionOptional.get();
+ Optional<IngestPosition<?>> position =
jobContext.getInitProgress().getIncrementalPosition(taskConfig.getDumperConfig().getDataSourceName());
+ if (position.isPresent()) {
+ return position.get();
}
}
String databaseType =
taskConfig.getJobConfig().getSourceDatabaseType();
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorker.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorker.java
index ac58354d882..7e33e163415 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorker.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorker.java
@@ -171,9 +171,9 @@ public final class RuleAlteredJobWorker {
log.warn("There is uncompleted job with the same database name,
please handle it first, current job will be ignored");
return;
}
- Optional<RuleAlteredJobConfiguration> jobConfigOptional =
createJobConfig(event);
- if (jobConfigOptional.isPresent()) {
-
RuleAlteredJobAPIFactory.getInstance().start(jobConfigOptional.get());
+ Optional<RuleAlteredJobConfiguration> jobConfig =
createJobConfig(event);
+ if (jobConfig.isPresent()) {
+ RuleAlteredJobAPIFactory.getInstance().start(jobConfig.get());
} else {
log.info("Switch rule configuration immediately.");
ScalingTaskFinishedEvent taskFinishedEvent = new
ScalingTaskFinishedEvent(event.getDatabaseName(), event.getActiveVersion(),
event.getNewVersion());
diff --git
a/shardingsphere-mode/shardingsphere-mode-core/src/test/java/org/apache/shardingsphere/mode/metadata/persist/service/SchemaMetaDataPersistServiceTest.java
b/shardingsphere-mode/shardingsphere-mode-core/src/test/java/org/apache/shardingsphere/mode/metadata/persist/service/SchemaMetaDataPersistServiceTest.java
index 701131efe96..55f60046365 100644
---
a/shardingsphere-mode/shardingsphere-mode-core/src/test/java/org/apache/shardingsphere/mode/metadata/persist/service/SchemaMetaDataPersistServiceTest.java
+++
b/shardingsphere-mode/shardingsphere-mode-core/src/test/java/org/apache/shardingsphere/mode/metadata/persist/service/SchemaMetaDataPersistServiceTest.java
@@ -79,15 +79,14 @@ public final class SchemaMetaDataPersistServiceTest {
SchemaMetaDataPersistService schemaMetaDataPersistService = new
SchemaMetaDataPersistService(repository);
when(repository.getChildrenKeys("/metadata/foo_db/schemas/foo_schema/tables")).thenReturn(Collections.singletonList("t_order"));
when(repository.get("/metadata/foo_db/schemas/foo_schema/tables/t_order")).thenReturn(readYAML());
- Optional<ShardingSphereSchema> schemaOptional =
schemaMetaDataPersistService.load("foo_db", "foo_schema");
- assertTrue(schemaOptional.isPresent());
+ Optional<ShardingSphereSchema> schema =
schemaMetaDataPersistService.load("foo_db", "foo_schema");
+ assertTrue(schema.isPresent());
Optional<ShardingSphereSchema> empty =
schemaMetaDataPersistService.load("test", "test");
assertThat(empty, is(Optional.empty()));
- ShardingSphereSchema schema = schemaOptional.get();
- assertThat(schema.getAllTableNames(),
is(Collections.singleton("t_order")));
- assertThat(schema.get("t_order").getIndexes().keySet(),
is(Collections.singleton("primary")));
- assertThat(schema.getAllColumnNames("t_order").size(), is(1));
- assertThat(schema.get("t_order").getColumns().keySet(),
is(Collections.singleton("id")));
+ assertThat(schema.get().getAllTableNames(),
is(Collections.singleton("t_order")));
+ assertThat(schema.get().get("t_order").getIndexes().keySet(),
is(Collections.singleton("primary")));
+ assertThat(schema.get().getAllColumnNames("t_order").size(), is(1));
+ assertThat(schema.get().get("t_order").getColumns().keySet(),
is(Collections.singleton("id")));
}
@Test
diff --git
a/shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/distsql/ral/common/queryable/ShowAuthorityRuleHandler.java
b/shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/distsql/ral/common/queryable/ShowAuthorityRuleHandler.java
index 47170279b4e..8152852a287 100644
---
a/shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/distsql/ral/common/queryable/ShowAuthorityRuleHandler.java
+++
b/shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/distsql/ral/common/queryable/ShowAuthorityRuleHandler.java
@@ -51,10 +51,7 @@ public final class ShowAuthorityRuleHandler extends
QueryableRALBackendHandler<S
protected Collection<List<Object>> getRows(final ContextManager
contextManager) {
Optional<AuthorityRuleConfiguration> authorityRuleConfig =
ProxyContext.getInstance().getContextManager()
.getMetaDataContexts().getGlobalRuleMetaData().findRuleConfigurations(AuthorityRuleConfiguration.class).stream().findFirst();
- if (!authorityRuleConfig.isPresent()) {
- return Collections.emptyList();
- }
- return Collections.singleton(getRow(authorityRuleConfig.get()));
+ return authorityRuleConfig.isPresent() ?
Collections.singleton(getRow(authorityRuleConfig.get())) :
Collections.emptyList();
}
private List<Object> getRow(final AuthorityRuleConfiguration
authorityRuleConfig) {
diff --git
a/shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/distsql/ral/common/queryable/ShowSQLParserRuleHandler.java
b/shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/distsql/ral/common/queryable/ShowSQLParserRuleHandler.java
index 6f817a5670e..45c79436d22 100644
---
a/shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/distsql/ral/common/queryable/ShowSQLParserRuleHandler.java
+++
b/shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/distsql/ral/common/queryable/ShowSQLParserRuleHandler.java
@@ -51,18 +51,16 @@ public final class ShowSQLParserRuleHandler extends
QueryableRALBackendHandler<S
@Override
protected Collection<List<Object>> getRows(final ContextManager
contextManager) {
- Optional<SQLParserRuleConfiguration>
sqlParserRuleConfigurationOptional =
ProxyContext.getInstance().getContextManager()
+ Optional<SQLParserRuleConfiguration> sqlParserRuleConfig =
ProxyContext.getInstance().getContextManager()
.getMetaDataContexts().getGlobalRuleMetaData().findRuleConfigurations(SQLParserRuleConfiguration.class).stream().findAny();
- if (!sqlParserRuleConfigurationOptional.isPresent()) {
- return Collections.emptyList();
- }
- SQLParserRuleConfiguration sqlParserRuleConfig =
sqlParserRuleConfigurationOptional.get();
- List<Object> row = new LinkedList<>();
-
row.add(String.valueOf(sqlParserRuleConfig.isSqlCommentParseEnabled()));
- row.add(GSON.toJson(sqlParserRuleConfig.getParseTreeCache()));
- row.add(GSON.toJson(sqlParserRuleConfig.getSqlStatementCache()));
- Collection<List<Object>> result = new LinkedList<>();
- result.add(row);
+ return sqlParserRuleConfig.isPresent() ?
Collections.singleton(getRow(sqlParserRuleConfig.get())) :
Collections.emptyList();
+ }
+
+ private List<Object> getRow(final SQLParserRuleConfiguration
sqlParserRuleConfig) {
+ List<Object> result = new LinkedList<>();
+
result.add(String.valueOf(sqlParserRuleConfig.isSqlCommentParseEnabled()));
+ result.add(GSON.toJson(sqlParserRuleConfig.getParseTreeCache()));
+ result.add(GSON.toJson(sqlParserRuleConfig.getSqlStatementCache()));
return result;
}
}
diff --git
a/shardingsphere-proxy/shardingsphere-proxy-backend/src/test/java/org/apache/shardingsphere/proxy/backend/text/admin/mysql/MySQLAdminExecutorFactoryTest.java
b/shardingsphere-proxy/shardingsphere-proxy-backend/src/test/java/org/apache/shardingsphere/proxy/backend/text/admin/mysql/MySQLAdminExecutorFactoryTest.java
index a26b25f6f1f..c40d2194b3b 100644
---
a/shardingsphere-proxy/shardingsphere-proxy-backend/src/test/java/org/apache/shardingsphere/proxy/backend/text/admin/mysql/MySQLAdminExecutorFactoryTest.java
+++
b/shardingsphere-proxy/shardingsphere-proxy-backend/src/test/java/org/apache/shardingsphere/proxy/backend/text/admin/mysql/MySQLAdminExecutorFactoryTest.java
@@ -37,35 +37,36 @@ import static org.mockito.Mockito.when;
public final class MySQLAdminExecutorFactoryTest {
- private final MySQLAdminExecutorCreator mySQLAdminExecutorFactory = new
MySQLAdminExecutorCreator();
-
+ @SuppressWarnings("rawtypes")
@Test
public void assertNewInstanceWithMySQLShowFunctionStatusStatement() {
MySQLShowFunctionStatusStatement statement =
mock(MySQLShowFunctionStatusStatement.class);
CommonSQLStatementContext statementContext =
mock(CommonSQLStatementContext.class);
when(statementContext.getSqlStatement()).thenReturn(statement);
- Optional<DatabaseAdminExecutor> executorOptional =
mySQLAdminExecutorFactory.create(statementContext);
- assertTrue(executorOptional.isPresent());
- assertThat(executorOptional.get(),
instanceOf(ShowFunctionStatusExecutor.class));
+ Optional<DatabaseAdminExecutor> executor = new
MySQLAdminExecutorCreator().create(statementContext);
+ assertTrue(executor.isPresent());
+ assertThat(executor.get(),
instanceOf(ShowFunctionStatusExecutor.class));
}
+ @SuppressWarnings("rawtypes")
@Test
public void assertNewInstanceWithMySQLShowProcedureStatusStatement() {
MySQLShowProcedureStatusStatement statement =
mock(MySQLShowProcedureStatusStatement.class);
CommonSQLStatementContext statementContext =
mock(CommonSQLStatementContext.class);
when(statementContext.getSqlStatement()).thenReturn(statement);
- Optional<DatabaseAdminExecutor> executorOptional =
mySQLAdminExecutorFactory.create(statementContext);
- assertTrue(executorOptional.isPresent());
- assertThat(executorOptional.get(),
instanceOf(ShowProcedureStatusExecutor.class));
+ Optional<DatabaseAdminExecutor> executor = new
MySQLAdminExecutorCreator().create(statementContext);
+ assertTrue(executor.isPresent());
+ assertThat(executor.get(),
instanceOf(ShowProcedureStatusExecutor.class));
}
+ @SuppressWarnings("rawtypes")
@Test
public void assertNewInstanceWithMySQLShowTablesStatement() {
MySQLShowTablesStatement statement =
mock(MySQLShowTablesStatement.class);
CommonSQLStatementContext statementContext =
mock(CommonSQLStatementContext.class);
when(statementContext.getSqlStatement()).thenReturn(statement);
- Optional<DatabaseAdminExecutor> executorOptional =
mySQLAdminExecutorFactory.create(statementContext);
- assertTrue(executorOptional.isPresent());
- assertThat(executorOptional.get(),
instanceOf(ShowTablesExecutor.class));
+ Optional<DatabaseAdminExecutor> executor = new
MySQLAdminExecutorCreator().create(statementContext);
+ assertTrue(executor.isPresent());
+ assertThat(executor.get(), instanceOf(ShowTablesExecutor.class));
}
}
diff --git
a/shardingsphere-sql-parser/shardingsphere-sql-parser-dialect/shardingsphere-sql-parser-opengauss/src/main/java/org/apache/shardingsphere/sql/parser/opengauss/visitor/statement/impl/OpenGaussDCLStatementSQLVisitor.java
b/shardingsphere-sql-parser/shardingsphere-sql-parser-dialect/shardingsphere-sql-parser-opengauss/src/main/java/org/apache/shardingsphere/sql/parser/opengauss/visitor/statement/impl/OpenGaussDCLStatementSQLVisitor.java
index db8cee4b4c5..18a9e5a3347 100644
---
a/shardingsphere-sql-parser/shardingsphere-sql-parser-dialect/shardingsphere-sql-parser-opengauss/src/main/java/org/apache/shardingsphere/sql/parser/opengauss/visitor/statement/impl/OpenGaussDCLStatementSQLVisitor.java
+++
b/shardingsphere-sql-parser/shardingsphere-sql-parser-dialect/shardingsphere-sql-parser-opengauss/src/main/java/org/apache/shardingsphere/sql/parser/opengauss/visitor/statement/impl/OpenGaussDCLStatementSQLVisitor.java
@@ -58,16 +58,16 @@ public final class OpenGaussDCLStatementSQLVisitor extends
OpenGaussStatementSQL
@Override
public ASTNode visitGrant(final GrantContext ctx) {
OpenGaussGrantStatement result = new OpenGaussGrantStatement();
- Optional<Collection<SimpleTableSegment>> tableSegmentOptional = null
== ctx.privilegeClause() ? Optional.empty() :
getTableFromPrivilegeClause(ctx.privilegeClause());
- tableSegmentOptional.ifPresent(optional ->
result.getTables().addAll(optional));
+ Optional<Collection<SimpleTableSegment>> tableSegment = null ==
ctx.privilegeClause() ? Optional.empty() :
getTableFromPrivilegeClause(ctx.privilegeClause());
+ tableSegment.ifPresent(optional ->
result.getTables().addAll(optional));
return result;
}
@Override
public ASTNode visitRevoke(final RevokeContext ctx) {
OpenGaussRevokeStatement result = new OpenGaussRevokeStatement();
- Optional<Collection<SimpleTableSegment>> tableSegmentOptional = null
== ctx.privilegeClause() ? Optional.empty() :
getTableFromPrivilegeClause(ctx.privilegeClause());
- tableSegmentOptional.ifPresent(optional ->
result.getTables().addAll(optional));
+ Optional<Collection<SimpleTableSegment>> tableSegment = null ==
ctx.privilegeClause() ? Optional.empty() :
getTableFromPrivilegeClause(ctx.privilegeClause());
+ tableSegment.ifPresent(optional ->
result.getTables().addAll(optional));
return result;
}
diff --git
a/shardingsphere-sql-parser/shardingsphere-sql-parser-dialect/shardingsphere-sql-parser-postgresql/src/main/java/org/apache/shardingsphere/sql/parser/postgresql/visitor/statement/impl/PostgreSQLDCLStatementSQLVisitor.java
b/shardingsphere-sql-parser/shardingsphere-sql-parser-dialect/shardingsphere-sql-parser-postgresql/src/main/java/org/apache/shardingsphere/sql/parser/postgresql/visitor/statement/impl/PostgreSQLDCLStatementSQLVisitor.java
index f9d9ddae8cc..ae118197ef1 100644
---
a/shardingsphere-sql-parser/shardingsphere-sql-parser-dialect/shardingsphere-sql-parser-postgresql/src/main/java/org/apache/shardingsphere/sql/parser/postgresql/visitor/statement/impl/PostgreSQLDCLStatementSQLVisitor.java
+++
b/shardingsphere-sql-parser/shardingsphere-sql-parser-dialect/shardingsphere-sql-parser-postgresql/src/main/java/org/apache/shardingsphere/sql/parser/postgresql/visitor/statement/impl/PostgreSQLDCLStatementSQLVisitor.java
@@ -58,16 +58,16 @@ public final class PostgreSQLDCLStatementSQLVisitor extends
PostgreSQLStatementS
@Override
public ASTNode visitGrant(final GrantContext ctx) {
PostgreSQLGrantStatement result = new PostgreSQLGrantStatement();
- Optional<Collection<SimpleTableSegment>> tableSegmentOptional = null
== ctx.privilegeClause() ? Optional.empty() :
getTableFromPrivilegeClause(ctx.privilegeClause());
- tableSegmentOptional.ifPresent(optional ->
result.getTables().addAll(optional));
+ Optional<Collection<SimpleTableSegment>> tableSegment = null ==
ctx.privilegeClause() ? Optional.empty() :
getTableFromPrivilegeClause(ctx.privilegeClause());
+ tableSegment.ifPresent(optional ->
result.getTables().addAll(optional));
return result;
}
@Override
public ASTNode visitRevoke(final RevokeContext ctx) {
PostgreSQLRevokeStatement result = new PostgreSQLRevokeStatement();
- Optional<Collection<SimpleTableSegment>> tableSegmentOptional = null
== ctx.privilegeClause() ? Optional.empty() :
getTableFromPrivilegeClause(ctx.privilegeClause());
- tableSegmentOptional.ifPresent(optional ->
result.getTables().addAll(optional));
+ Optional<Collection<SimpleTableSegment>> tableSegment = null ==
ctx.privilegeClause() ? Optional.empty() :
getTableFromPrivilegeClause(ctx.privilegeClause());
+ tableSegment.ifPresent(optional ->
result.getTables().addAll(optional));
return result;
}
diff --git
a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/JobProgressTest.java
b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/JobProgressTest.java
index f49b80c8f2e..784face4584 100644
---
a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/JobProgressTest.java
+++
b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/JobProgressTest.java
@@ -52,9 +52,9 @@ public final class JobProgressTest {
@Test
public void assertGetIncrementalPosition() {
JobProgress jobProgress =
getJobProgress(ConfigurationFileUtil.readFile("job-progress.yaml"));
- Optional<IngestPosition<?>> positionOptional =
jobProgress.getIncrementalPosition("ds0");
- assertTrue(positionOptional.isPresent());
- assertThat(positionOptional.get(),
instanceOf(PlaceholderPosition.class));
+ Optional<IngestPosition<?>> position =
jobProgress.getIncrementalPosition("ds0");
+ assertTrue(position.isPresent());
+ assertThat(position.get(), instanceOf(PlaceholderPosition.class));
}
@Test
