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

fix: adds order by timestamp #4316

Merged
merged 2 commits into from
Jul 2, 2024
Merged

fix: adds order by timestamp #4316

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
9bd2470
fix: adds order by timestamp + tests in state machines
wolf4ood Jul 1, 2024
9957c93
pr remarks
wolf4ood Jul 2, 2024
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
26 changes: 14 additions & 12 deletions .../main/java/org/eclipse/edc/connector/policy/monitor/manager/PolicyMonitorManagerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,12 +52,6 @@ private PolicyMonitorManagerImpl() {

}

@Override
protected StateMachineManager.Builder configureStateMachineManager(StateMachineManager.Builder builder) {
return builder
.processor(processEntriesInState(STARTED, this::processMonitoring));
}

@Override
public void startMonitoring(String transferProcessId, String contractId) {
var entry = PolicyMonitorEntry.Builder.newInstance()
Expand All @@ -71,6 +65,12 @@ public void startMonitoring(String transferProcessId, String contractId) {
update(entry);
}

@Override
protected StateMachineManager.Builder configureStateMachineManager(StateMachineManager.Builder builder) {
return builder
.processor(processEntriesInState(STARTED, this::processMonitoring));
}

private boolean processMonitoring(PolicyMonitorEntry entry) {
var transferProcess = transferProcessService.findById(entry.getId());
if (transferProcess == null) {
Expand Down Expand Up @@ -110,8 +110,10 @@ private boolean processMonitoring(PolicyMonitorEntry entry) {
}
}

breakLease(entry);
return true;
// we update the state timestamp ensure fairness on polling on `STARTED` state
// the lease will be broken in `onNotProcessed`
entry.updateStateTimestamp();
return false;
}

private Processor processEntriesInState(PolicyMonitorEntryStates state, Function<PolicyMonitorEntry, Boolean> function) {
Expand All @@ -125,14 +127,14 @@ private Processor processEntriesInState(PolicyMonitorEntryStates state, Function
public static class Builder
extends AbstractStateEntityManager.Builder<PolicyMonitorEntry, PolicyMonitorStore, PolicyMonitorManagerImpl, Builder> {

public static Builder newInstance() {
return new Builder();
}

private Builder() {
super(new PolicyMonitorManagerImpl());
}

public static Builder newInstance() {
return new Builder();
}

public Builder contractAgreementService(ContractAgreementService contractAgreementService) {
manager.contractAgreementService = contractAgreementService;
return this;
Expand Down
4 changes: 3 additions & 1 deletion ...t/java/org/eclipse/edc/connector/policy/monitor/manager/PolicyMonitorManagerImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,6 +149,8 @@ void started_shouldDoNothing_whenPolicyIsValid() {
.state(STARTED.code())
.build();
var policy = Policy.Builder.newInstance().build();

var stateTimestamp = entry.getStateTimestamp();
when(store.nextNotLeased(anyInt(), stateIs(STARTED.code()))).thenReturn(List.of(entry)).thenReturn(emptyList());
when(transferProcessService.findById(entry.getId()))
.thenReturn(TransferProcess.Builder.newInstance().state(TransferProcessStates.STARTED.code()).build());
Expand All @@ -159,7 +161,7 @@ void started_shouldDoNothing_whenPolicyIsValid() {

await().untilAsserted(() -> {
verify(transferProcessService, never()).terminate(any());
verify(store).save(argThat(it -> it.getState() == STARTED.code()));
verify(store).save(argThat(it -> it.getState() == STARTED.code() && stateTimestamp < it.getStateTimestamp()));
});
}

Expand Down
4 changes: 4 additions & 0 deletions extensions/control-plane/store/sql/contract-negotiation-store-sql/docs/schema.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,3 +80,7 @@ CREATE UNIQUE INDEX IF NOT EXISTS contract_negotiation_id_uindex

CREATE UNIQUE INDEX IF NOT EXISTS contract_agreement_id_uindex
ON edc_contract_agreement (agr_id);


-- This will help to identify states that need to be transitioned without a table scan when the entries grow
CREATE INDEX IF NOT EXISTS contract_negotiation_state ON edc_contract_negotiation (state,state_timestamp);
94 changes: 47 additions & 47 deletions ...nnector/controlplane/store/sql/contractnegotiation/store/SqlContractNegotiationStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,21 +79,39 @@ public SqlContractNegotiationStore(DataSourceRegistry dataSourceRegistry, String
}

@Override
public @Nullable ContractNegotiation findForCorrelationId(String correlationId) {
public @NotNull List<ContractNegotiation> nextNotLeased(int max, Criterion... criteria) {
return transactionContext.execute(() -> {
// utilize the generic query api
var query = correlationIdQuerySpec(correlationId);
try (var stream = queryNegotiations(query)) {
return single(stream.collect(toList()));
var filter = Arrays.stream(criteria).toList();
var querySpec = QuerySpec.Builder.newInstance().filter(filter).sortField("stateTimestamp").limit(max).build();
var statement = statements.createNegotiationsQuery(querySpec)
.addWhereClause(statements.getNotLeasedFilter(), clock.millis());

try (
var connection = getConnection();
var stream = queryExecutor.query(getConnection(), true, contractNegotiationWithAgreementMapper(connection), statement.getQueryAsString(), statement.getParameters())
) {
var negotiations = stream.collect(toList());
negotiations.forEach(cn -> leaseContext.withConnection(connection).acquireLease(cn.getId()));
return negotiations;
} catch (SQLException e) {
throw new EdcPersistenceException(e);
}
});
}

@Override
public @Nullable ContractAgreement findContractAgreement(String contractId) {
public StoreResult<ContractNegotiation> findByIdAndLease(String id) {
return transactionContext.execute(() -> {
try (var connection = getConnection()) {
return findContractAgreementInternal(connection, contractId);
var entity = findInternal(connection, id);
if (entity == null) {
return StoreResult.notFound(format("ContractNegotiation %s not found", id));
}

leaseContext.withConnection(connection).acquireLease(id);
return StoreResult.success(entity);
} catch (IllegalStateException e) {
return StoreResult.alreadyLeased(format("ContractNegotiation %s is already leased", id));
} catch (SQLException e) {
throw new EdcPersistenceException(e);
}
Expand All @@ -119,6 +137,28 @@ public void save(ContractNegotiation negotiation) {

}

@Override
public @Nullable ContractNegotiation findForCorrelationId(String correlationId) {
return transactionContext.execute(() -> {
// utilize the generic query api
var query = correlationIdQuerySpec(correlationId);
try (var stream = queryNegotiations(query)) {
return single(stream.collect(toList()));
}
});
}

@Override
public @Nullable ContractAgreement findContractAgreement(String contractId) {
return transactionContext.execute(() -> {
try (var connection = getConnection()) {
return findContractAgreementInternal(connection, contractId);
} catch (SQLException e) {
throw new EdcPersistenceException(e);
}
});
}

@Override
public void delete(String negotiationId) {
transactionContext.execute(() -> {
Expand Down Expand Up @@ -171,46 +211,6 @@ public void delete(String negotiationId) {
});
}

@Override
public @NotNull List<ContractNegotiation> nextNotLeased(int max, Criterion... criteria) {
return transactionContext.execute(() -> {
var filter = Arrays.stream(criteria).toList();
var querySpec = QuerySpec.Builder.newInstance().filter(filter).limit(max).build();
var statement = statements.createNegotiationsQuery(querySpec)
.addWhereClause(statements.getNotLeasedFilter(), clock.millis());

try (
var connection = getConnection();
var stream = queryExecutor.query(getConnection(), true, contractNegotiationWithAgreementMapper(connection), statement.getQueryAsString(), statement.getParameters())
) {
var negotiations = stream.collect(toList());
negotiations.forEach(cn -> leaseContext.withConnection(connection).acquireLease(cn.getId()));
return negotiations;
} catch (SQLException e) {
throw new EdcPersistenceException(e);
}
});
}

@Override
public StoreResult<ContractNegotiation> findByIdAndLease(String id) {
return transactionContext.execute(() -> {
try (var connection = getConnection()) {
var entity = findInternal(connection, id);
if (entity == null) {
return StoreResult.notFound(format("ContractNegotiation %s not found", id));
}

leaseContext.withConnection(connection).acquireLease(id);
return StoreResult.success(entity);
} catch (IllegalStateException e) {
return StoreResult.alreadyLeased(format("ContractNegotiation %s is already leased", id));
} catch (SQLException e) {
throw new EdcPersistenceException(e);
}
});
}

@Override
public StoreResult<ContractNegotiation> findByCorrelationIdAndLease(String correlationId) {
return transactionContext.execute(() -> {
Expand Down
3 changes: 3 additions & 0 deletions extensions/control-plane/store/sql/transfer-process-store-sql/docs/schema.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,3 +65,6 @@ CREATE UNIQUE INDEX IF NOT EXISTS transfer_process_id_uindex

CREATE UNIQUE INDEX IF NOT EXISTS lease_lease_id_uindex
ON edc_lease (lease_id);

-- This will help to identify states that need to be transitioned without a table scan when the entries grow
CREATE INDEX IF NOT EXISTS transfer_process_state ON edc_transfer_process (state,state_time_stamp);
72 changes: 36 additions & 36 deletions ...e/edc/connector/controlplane/store/sql/transferprocess/store/SqlTransferProcessStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,11 +70,22 @@ public SqlTransferProcessStore(DataSourceRegistry dataSourceRegistry, String dat
leaseContext = SqlLeaseContextBuilder.with(transactionContext, leaseHolderName, statements, clock, queryExecutor);
}

@Override
public @Nullable TransferProcess findById(String id) {
return transactionContext.execute(() -> {
try (var connection = getConnection()) {
return findByIdInternal(connection, id);
} catch (SQLException e) {
throw new EdcPersistenceException(e);
}
});
}

@Override
public @NotNull List<TransferProcess> nextNotLeased(int max, Criterion... criteria) {
return transactionContext.execute(() -> {
var filter = Arrays.stream(criteria).collect(toList());
var querySpec = QuerySpec.Builder.newInstance().filter(filter).limit(max).build();
var querySpec = QuerySpec.Builder.newInstance().filter(filter).sortField("stateTimestamp").limit(max).build();
var statement = statements.createQuery(querySpec)
.addWhereClause(statements.getNotLeasedFilter(), clock.millis());

Expand Down Expand Up @@ -110,30 +121,6 @@ public StoreResult<TransferProcess> findByIdAndLease(String id) {
});
}

@Override
public StoreResult<TransferProcess> findByCorrelationIdAndLease(String correlationId) {
return transactionContext.execute(() -> {
var query = correlationIdQuerySpec(correlationId);

try (
var connection = getConnection();
var stream = executeQuery(connection, query)
) {
var entity = stream.findFirst().orElse(null);
if (entity == null) {
return StoreResult.notFound(format("TransferProcess with correlationId %s not found", correlationId));
}

leaseContext.withConnection(connection).acquireLease(entity.getId());
return StoreResult.success(entity);
} catch (IllegalStateException e) {
return StoreResult.alreadyLeased(format("TransferProcess with correlationId %s is already leased", correlationId));
} catch (SQLException e) {
throw new EdcPersistenceException(e);
}
});
}

@Override
public void save(TransferProcess entity) {
Objects.requireNonNull(entity.getId(), "TransferProcesses must have an ID!");
Expand All @@ -152,17 +139,6 @@ public void save(TransferProcess entity) {
});
}

@Override
public @Nullable TransferProcess findById(String id) {
return transactionContext.execute(() -> {
try (var connection = getConnection()) {
return findByIdInternal(connection, id);
} catch (SQLException e) {
throw new EdcPersistenceException(e);
}
});
}

@Override
public @Nullable TransferProcess findForCorrelationId(String correlationId) {
return transactionContext.execute(() -> {
Expand Down Expand Up @@ -207,6 +183,30 @@ public Stream<TransferProcess> findAll(QuerySpec querySpec) {
});
}

@Override
public StoreResult<TransferProcess> findByCorrelationIdAndLease(String correlationId) {
return transactionContext.execute(() -> {
var query = correlationIdQuerySpec(correlationId);

try (
var connection = getConnection();
var stream = executeQuery(connection, query)
) {
var entity = stream.findFirst().orElse(null);
if (entity == null) {
return StoreResult.notFound(format("TransferProcess with correlationId %s not found", correlationId));
}

leaseContext.withConnection(connection).acquireLease(entity.getId());
return StoreResult.success(entity);
} catch (IllegalStateException e) {
return StoreResult.alreadyLeased(format("TransferProcess with correlationId %s is already leased", correlationId));
} catch (SQLException e) {
throw new EdcPersistenceException(e);
}
});
}

private QuerySpec correlationIdQuerySpec(String correlationId) {
var criterion = criterion("correlationId", "=", correlationId);
return QuerySpec.Builder.newInstance().filter(criterion).build();
Expand Down
3 changes: 3 additions & 0 deletions extensions/data-plane/store/sql/data-plane-store-sql/docs/schema.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,3 +38,6 @@ COMMENT ON COLUMN edc_data_plane.trace_context IS 'Java Map serialized as JSON';
COMMENT ON COLUMN edc_data_plane.source IS 'DataAddress serialized as JSON';
COMMENT ON COLUMN edc_data_plane.destination IS 'DataAddress serialized as JSON';
COMMENT ON COLUMN edc_data_plane.properties IS 'Java Map serialized as JSON';

-- This will help to identify states that need to be transitioned without a table scan when the entries grow
CREATE INDEX IF NOT EXISTS data_plane_state ON edc_data_plane (state,state_time_stamp);
2 changes: 1 addition & 1 deletion ...re-sql/src/main/java/org/eclipse/edc/connector/dataplane/store/sql/SqlDataPlaneStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ public SqlDataPlaneStore(DataSourceRegistry dataSourceRegistry, String dataSourc
public @NotNull List<DataFlow> nextNotLeased(int max, Criterion... criteria) {
return transactionContext.execute(() -> {
var filter = Arrays.stream(criteria).collect(toList());
var querySpec = QuerySpec.Builder.newInstance().filter(filter).limit(max).build();
var querySpec = QuerySpec.Builder.newInstance().filter(filter).sortField("stateTimestamp").limit(max).build();
var statement = statements.createQuery(querySpec)
.addWhereClause(statements.getNotLeasedFilter(), clock.millis());

Expand Down
4 changes: 4 additions & 0 deletions extensions/policy-monitor/store/sql/policy-monitor-store-sql/docs/schema.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,3 +30,7 @@ CREATE TABLE IF NOT EXISTS edc_policy_monitor
properties JSON,
contract_id VARCHAR
);


-- This will help to identify states that need to be transitioned without a table scan when the entries grow
CREATE INDEX IF NOT EXISTS policy_monitor_state ON edc_policy_monitor (state,state_time_stamp);
2 changes: 1 addition & 1 deletion ...c/main/java/org/eclipse/edc/connector/policy/monitor/store/sql/SqlPolicyMonitorStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ public SqlPolicyMonitorStore(DataSourceRegistry dataSourceRegistry, String dataS
public @NotNull List<PolicyMonitorEntry> nextNotLeased(int max, Criterion... criteria) {
return transactionContext.execute(() -> {
var filter = Arrays.stream(criteria).collect(toList());
var querySpec = QuerySpec.Builder.newInstance().filter(filter).limit(max).build();
var querySpec = QuerySpec.Builder.newInstance().filter(filter).sortField("stateTimestamp").limit(max).build();
var statement = statements.createQuery(querySpec)
.addWhereClause(statements.getNotLeasedFilter(), clock.millis());

Expand Down
Loading
Loading