You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by mb...@apache.org on 2023/08/01 03:48:43 UTC
[asterixdb] branch master updated (1cb3eed1da -> 89c10eb2b1)
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
from 1cb3eed1da [ASTERIXDB-3232][COMP] cleanup and simplification
new 677f1ae697 [NO ISSUE][HYR][HTTP] Prevent Netty buffer leaks on close() errors
new 767596bb26 [NO ISSUE][*DB] Provide ability for IDatasetRebalanceCallback to decline a rebalance
new 8b67de4caa [NO ISSUE][*DB] Update dependencies
new 1e367fd1fc [ASTERIXDB-3216][NET] Always resolve NC address on node registration
new 2f56ed9d6d [ASTERIXDB-3219][REPL] Add timeout to log replication
new cc7d89ccbd [ASTERIXDB-3220][REPL] Allow waiting for IO on specific dataset partition
new 5f076a8347 [ASTERIXDB-3221][REPL] Use IO scheduler for replication ops
new 9912283399 [NO ISSUE][OTH] Fix log message
new c284d360d3 [NO ISSUE][OTH] Move page size storage config to common config
new 0dded85def [ASTERIXDB-3216][NET] Always resolve NC address on node registration
new e64d046289 [ASTERIXDB-3226][STO] Warn on missing primary index during flush
new 04b1dd4f41 Merge branch 'gerrit/stabilization-667a908755' into neo
new cf40ec4341 [NO ISSUE] Update dependencies
new 84c47cf88a Merge branch 'gerrit/stabilization-5949a1cb71' into 'gerrit/mad-hatter'
new 1e5910e2df Merge branch 'gerrit/mad-hatter' into 'gerrit/cheshire-cat'
new c1a1742a1f Merge branch 'gerrit/cheshire-cat' into 'gerrit/neo'
new 641462e651 [NO ISSUE][*DB] Enable ability to configure RMI bind address
new 47e2613e0c Merge branch 'gerrit/stabilization-667a908755' into 'gerrit/neo'
new 0f3fb3b0b6 Merge branch 'gerrit/neo' into 'gerrit/trinity'
new 89c10eb2b1 Merge branch 'gerrit/trinity' into 'master'
The 20 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../http/server/NCQueryCancellationServlet.java | 3 +
.../api/http/server/NCQueryServiceServlet.java | 3 +-
.../app/active/ActiveEntityEventsListener.java | 10 +-
.../app/active/ActiveNotificationHandler.java | 52 +++++-----
.../asterix/app/message/CancelQueryRequest.java | 8 +-
.../message/ExecuteStatementRequestMessage.java | 4 +-
.../org/apache/asterix/app/nc/HaltCallback.java | 4 +-
.../apache/asterix/app/nc/NCAppRuntimeContext.java | 16 +---
.../asterix/app/translator/QueryTranslator.java | 4 +
.../rebalance/IDatasetRebalanceCallback.java | 8 +-
.../rebalance/NoOpDatasetRebalanceCallback.java | 3 +-
.../org/apache/asterix/utils/RebalanceUtil.java | 42 +++++---
.../{cc_no_cbo.conf => cc-cloud-storage.conf} | 4 +-
.../src/main/resources/cc_static_partitioning.conf | 6 +-
.../asterix/test/common/ResultExtractor.java | 4 +-
.../asterix-app/src/test/resources/cc-analyze.conf | 2 +-
.../src/test/resources/cc-cloud-storage.conf | 7 --
.../common/api/IDatasetLifecycleManager.java | 6 +-
.../common/context/BaseOperationTracker.java | 8 +-
.../apache/asterix/common/context/DatasetInfo.java | 31 +++++-
.../common/context/DatasetLifecycleManager.java | 4 +-
.../context/PrimaryIndexOperationTracker.java | 11 +--
.../ioopcallbacks/LSMIOOperationCallback.java | 6 +-
.../ioopcallbacks/LSMIOOperationCallbackTest.java | 7 ++
.../apache/asterix/metadata/RMIClientFactory.java | 13 ++-
.../apache/asterix/metadata/RMIServerFactory.java | 25 ++++-
.../metadata/bootstrap/AsterixStateProxy.java | 15 +--
.../asterix/metadata/entities/NodeGroup.java | 3 +-
.../apache/asterix/metadata/utils/DatasetUtil.java | 2 +-
.../replication/api/ReplicationDestination.java | 32 +++++++
.../management/IndexReplicationManager.java | 99 +++++++++----------
.../management/ReplicationOperation.java | 106 +++++++++++++++++++++
.../replication/sync/ReplicaSynchronizer.java | 2 +-
.../SecondaryIndexOperationTrackerFactory.java | 4 +-
.../service/logging/LogManagerWithReplication.java | 19 +++-
asterixdb/pom.xml | 4 +-
.../appended-resources/supplemental-models.xml | 97 +++++++++----------
....com_netty_netty_netty-4.1.94.Final_NOTICE.txt} | 0
...ent.com_xerial_snappy-java_v1.1.10.1_NOTICE.txt | 16 ++++
.../apache/hyracks/api/comm/NetworkAddress.java | 7 +-
.../api/network/INetworkSecurityConfig.java | 9 ++
.../org/apache/hyracks/api/util/InvokeUtil.java | 40 +++++++-
.../hyracks/control/cc/work/CancelJobWork.java | 5 +
.../cc/work/JobletCleanupNotificationWork.java | 7 +-
.../hyracks/control/cc/work/RegisterNodeWork.java | 6 +-
.../control/cc/work/WaitForJobCompletionWork.java | 5 +
.../hyracks/control/nc/NodeControllerService.java | 2 +-
.../hyracks/control/nc/work/AbortTasksWork.java | 5 +
.../hyracks/control/nc/work/CleanupJobletWork.java | 7 +-
.../hyracks/control/nc/work/StartTasksWork.java | 5 +
.../hyracks/http/server/ChunkedResponse.java | 57 +++++++----
.../ipc/security/NetworkSecurityConfig.java | 7 ++
.../impls/AbstractAsynchronousScheduler.java | 52 +++++++++-
.../{ThrowingAction.java => IOThrowingAction.java} | 8 +-
hyracks-fullstack/pom.xml | 25 ++++-
55 files changed, 685 insertions(+), 252 deletions(-)
copy asterixdb/asterix-app/src/main/resources/{cc_no_cbo.conf => cc-cloud-storage.conf} (96%)
create mode 100644 asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationOperation.java
rename asterixdb/src/main/licenses/content/{raw.githubusercontent.com_netty_netty_netty-4.1.87.Final_NOTICE.txt => raw.githubusercontent.com_netty_netty_netty-4.1.94.Final_NOTICE.txt} (100%)
create mode 100644 asterixdb/src/main/licenses/content/raw.githubusercontent.com_xerial_snappy-java_v1.1.10.1_NOTICE.txt
copy hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/{ThrowingAction.java => IOThrowingAction.java} (83%)
[asterixdb] 18/20: Merge branch 'gerrit/stabilization-667a908755' into 'gerrit/neo'
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 47e2613e0cf2b975c87c930325880251fd792c6b
Merge: c1a1742a1f 641462e651
Author: Michael Blow <mb...@apache.org>
AuthorDate: Thu Jul 27 21:20:55 2023 -0400
Merge branch 'gerrit/stabilization-667a908755' into 'gerrit/neo'
Change-Id: Id7219e74699c96a473b9bbe91ad30c0ae8d16de6
.../apache/asterix/app/nc/NCAppRuntimeContext.java | 16 ++++----------
.../apache/asterix/metadata/RMIClientFactory.java | 13 ++++++++++-
.../apache/asterix/metadata/RMIServerFactory.java | 25 +++++++++++++++++++---
.../metadata/bootstrap/AsterixStateProxy.java | 15 +++----------
.../api/network/INetworkSecurityConfig.java | 9 ++++++++
.../ipc/security/NetworkSecurityConfig.java | 7 ++++++
6 files changed, 57 insertions(+), 28 deletions(-)
[asterixdb] 09/20: [NO ISSUE] Update dependencies
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit cf40ec43410fb98533cfcb493f184f468a8f8d24
Author: Michael Blow <mi...@couchbase.com>
AuthorDate: Fri Jul 14 15:17:09 2023 -0400
[NO ISSUE] Update dependencies
- io.netty:netty-buffer ............... 4.1.87.Final -> 4.1.94.Final
- io.netty:netty-codec ................ 4.1.87.Final -> 4.1.94.Final
- io.netty:netty-codec-http ........... 4.1.87.Final -> 4.1.94.Final
- io.netty:netty-common ............... 4.1.87.Final -> 4.1.94.Final
- io.netty:netty-handler .............. 4.1.87.Final -> 4.1.94.Final
- io.netty:netty-transport ............ 4.1.87.Final -> 4.1.94.Final
Change-Id: I910f277f2e11df3cfe768312d032544c1a43e7a7
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17642
Reviewed-by: Ali Alsuliman <al...@gmail.com>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
---
...sercontent.com_netty_netty_netty-4.1.94.Final_NOTICE.txt} | 0
hyracks-fullstack/pom.xml | 12 ++++++------
2 files changed, 6 insertions(+), 6 deletions(-)
diff --git a/asterixdb/src/main/licenses/content/raw.githubusercontent.com_netty_netty_netty-4.1.87.Final_NOTICE.txt b/asterixdb/src/main/licenses/content/raw.githubusercontent.com_netty_netty_netty-4.1.94.Final_NOTICE.txt
similarity index 100%
rename from asterixdb/src/main/licenses/content/raw.githubusercontent.com_netty_netty_netty-4.1.87.Final_NOTICE.txt
rename to asterixdb/src/main/licenses/content/raw.githubusercontent.com_netty_netty_netty-4.1.94.Final_NOTICE.txt
diff --git a/hyracks-fullstack/pom.xml b/hyracks-fullstack/pom.xml
index 1b4d1f01cd..f3f75532e2 100644
--- a/hyracks-fullstack/pom.xml
+++ b/hyracks-fullstack/pom.xml
@@ -86,32 +86,32 @@
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-buffer</artifactId>
- <version>4.1.87.Final</version>
+ <version>4.1.94.Final</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-common</artifactId>
- <version>4.1.87.Final</version>
+ <version>4.1.94.Final</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-codec-http</artifactId>
- <version>4.1.87.Final</version>
+ <version>4.1.94.Final</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport</artifactId>
- <version>4.1.87.Final</version>
+ <version>4.1.94.Final</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-codec</artifactId>
- <version>4.1.87.Final</version>
+ <version>4.1.94.Final</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-handler</artifactId>
- <version>4.1.87.Final</version>
+ <version>4.1.94.Final</version>
</dependency>
<dependency>
<groupId>junit</groupId>
[asterixdb] 04/20: [ASTERIXDB-3216][NET] Always resolve NC address on node registration
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 1e367fd1fc74e4614b9f7c1eeea1c3155c2a9615
Author: Ali Alsuliman <al...@gmail.com>
AuthorDate: Tue Jul 4 21:30:31 2023 -0700
[ASTERIXDB-3216][NET] Always resolve NC address on node registration
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
Always resolve NC address on node registration.
- add some loggings
Change-Id: I37477de0a932439e301d8ff6f88de1355d612736
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17628
Reviewed-by: Michael Blow <mb...@apache.org>
Tested-by: Michael Blow <mb...@apache.org>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
---
.../http/server/NCQueryCancellationServlet.java | 3 ++
.../api/http/server/NCQueryServiceServlet.java | 3 +-
.../app/active/ActiveEntityEventsListener.java | 10 ++---
.../app/active/ActiveNotificationHandler.java | 52 +++++++++++-----------
.../asterix/app/message/CancelQueryRequest.java | 8 +++-
.../message/ExecuteStatementRequestMessage.java | 4 +-
.../asterix/app/translator/QueryTranslator.java | 4 ++
.../apache/hyracks/api/comm/NetworkAddress.java | 7 ++-
.../hyracks/control/cc/work/CancelJobWork.java | 5 +++
.../cc/work/JobletCleanupNotificationWork.java | 7 ++-
.../hyracks/control/cc/work/RegisterNodeWork.java | 6 ++-
.../control/cc/work/WaitForJobCompletionWork.java | 5 +++
.../hyracks/control/nc/NodeControllerService.java | 2 +-
.../hyracks/control/nc/work/AbortTasksWork.java | 5 +++
.../hyracks/control/nc/work/CleanupJobletWork.java | 7 ++-
.../hyracks/control/nc/work/StartTasksWork.java | 5 +++
16 files changed, 91 insertions(+), 42 deletions(-)
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryCancellationServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryCancellationServlet.java
index 41f2a53dc2..b2134dcfc6 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryCancellationServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryCancellationServlet.java
@@ -67,6 +67,9 @@ public class NCQueryCancellationServlet extends AbstractServlet {
CancelQueryRequest cancelQueryMessage = new CancelQueryRequest(serviceCtx.getNodeId(),
cancelQueryFuture.getFutureId(), uuid, clientContextId);
// TODO(mblow): multicc -- need to send cancellation to the correct cc
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Sending CancelQueryRequest with uuid:{}, clientContextID:{}", uuid, clientContextId);
+ }
messageBroker.sendMessageToPrimaryCC(cancelQueryMessage);
CancelQueryResponse cancelResponse =
(CancelQueryResponse) cancelQueryFuture.get(DEFAULT_NC_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryServiceServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryServiceServlet.java
index 1ec74055d0..9fa479c398 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryServiceServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryServiceServlet.java
@@ -165,7 +165,8 @@ public class NCQueryServiceServlet extends QueryServiceServlet {
CancelQueryRequest cancelQueryMessage =
new CancelQueryRequest(nodeId, cancelQueryFuture.getFutureId(), uuid, clientContextID);
// TODO(mblow): multicc -- need to send cancellation to the correct cc
- LOGGER.info("Cancelling query due to {}", exception.getClass().getSimpleName());
+ LOGGER.info("Cancelling query with uuid:{}, clientContextID:{} due to {}", uuid, clientContextID,
+ exception.getClass().getSimpleName());
messageBroker.sendMessageToPrimaryCC(cancelQueryMessage);
if (wait) {
cancelQueryFuture.get(ExecuteStatementRequestMessage.DEFAULT_QUERY_CANCELLATION_WAIT_MILLIS,
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
index ddd3d64682..3fd339e8df 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
@@ -195,12 +195,12 @@ public abstract class ActiveEntityEventsListener implements IActiveEntityControl
@SuppressWarnings("unchecked")
protected void finish(ActiveEvent event) throws HyracksDataException {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "the job {} finished", jobId);
+ LOGGER.log(level, "Active job {} finished", jobId);
}
JobId lastJobId = jobId;
if (numRegistered != numDeRegistered) {
LOGGER.log(Level.WARN,
- "the job {} finished with reported runtime registrations = {} and deregistrations = {}", jobId,
+ "Active job {} finished with reported runtime registrations = {} and deregistrations = {}", jobId,
numRegistered, numDeRegistered);
}
jobId = null;
@@ -208,7 +208,7 @@ public abstract class ActiveEntityEventsListener implements IActiveEntityControl
JobStatus jobStatus = status.getLeft();
List<Exception> exceptions = status.getRight();
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "The job finished with status: {}", jobStatus);
+ LOGGER.log(level, "Active job {} finished with status {}", lastJobId, jobStatus);
}
if (!jobSuccessfullyTerminated(jobStatus)) {
jobFailure = exceptions.isEmpty() ? new RuntimeDataException(ErrorCode.UNREPORTED_TASK_FAILURE_EXCEPTION)
@@ -372,7 +372,7 @@ public abstract class ActiveEntityEventsListener implements IActiveEntityControl
@Override
public synchronized void recover() {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Recover is called on " + entityId);
+ LOGGER.log(level, "Recover is called on {}", entityId);
}
if (retryPolicyFactory == NoRetryPolicyFactory.INSTANCE) {
LOGGER.log(level, "But it has no recovery policy, so it is set to permanent failure");
@@ -456,7 +456,7 @@ public abstract class ActiveEntityEventsListener implements IActiveEntityControl
try {
metadataProvider.getApplicationContext().getHcc().cancelJob(jobId);
} catch (Throwable th) {
- LOGGER.warn("Failed to cancel active job", th);
+ LOGGER.warn("Failed to cancel active job {}", jobId, th);
e.addSuppressed(th);
}
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java
index 284929f5ae..6b3581e66c 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java
@@ -66,22 +66,24 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
@Override
protected void handle(ActiveEvent event) {
- EntityId entityId = jobId2EntityId.get(event.getJobId());
+ JobId jobId = event.getJobId();
+ Kind eventKind = event.getEventKind();
+ EntityId entityId = jobId2EntityId.get(jobId);
if (entityId != null) {
IActiveEntityEventsListener listener = entityEventListeners.get(entityId);
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Next event is of type " + event.getEventKind());
+ LOGGER.log(level, "Next event is {} for job {}", eventKind, jobId);
}
- if (event.getEventKind() == Kind.JOB_FINISHED) {
- LOGGER.log(level, "Removing the job");
- jobId2EntityId.remove(event.getJobId());
+ if (eventKind == Kind.JOB_FINISHED) {
+ LOGGER.log(level, "Removing job {}", jobId);
+ jobId2EntityId.remove(jobId);
}
if (listener != null) {
LOGGER.log(level, "Notifying the listener");
listener.notify(event);
}
} else {
- LOGGER.log(Level.ERROR, "Entity not found for received message for job " + event.getJobId());
+ LOGGER.log(Level.ERROR, "Entity not found for event {} for job {}", eventKind, jobId);
}
}
@@ -90,45 +92,43 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
@Override
public void notifyJobCreation(JobId jobId, JobSpecification jobSpecification) throws HyracksDataException {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level,
- "notifyJobCreation(JobId jobId, JobSpecification jobSpecification) was called with jobId = "
- + jobId);
+ LOGGER.log(level, "notifyJobCreation was called for job {}", jobId);
}
Object property = jobSpecification.getProperty(ACTIVE_ENTITY_PROPERTY_NAME);
if (!(property instanceof EntityId)) {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Job is not of type active job. property found to be: " + property);
+ LOGGER.log(level, "Job {} is not of type active job. property found to be {}", jobId, property);
}
return;
}
EntityId entityId = (EntityId) property;
monitorJob(jobId, entityId);
boolean found = jobId2EntityId.get(jobId) != null;
- LOGGER.log(level, "Job was found to be: " + (found ? "Active" : "Inactive"));
+ LOGGER.log(level, "Job {} was found to be {}", jobId, (found ? "Active" : "Inactive"));
add(new ActiveEvent(jobId, Kind.JOB_CREATED, entityId, jobSpecification));
}
private synchronized void monitorJob(JobId jobId, EntityId entityId) {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "monitorJob(JobId jobId, ActiveJob activeJob) called with job id: " + jobId);
+ LOGGER.log(level, "monitorJob was called for job {}", jobId);
}
boolean found = jobId2EntityId.get(jobId) != null;
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Job was found to be: " + (found ? "Active" : "Inactive"));
+ LOGGER.log(level, "Job {} was found to be {}", jobId, (found ? "Active" : "Inactive"));
}
if (entityEventListeners.containsKey(entityId)) {
if (jobId2EntityId.containsKey(jobId)) {
if (LOGGER.isErrorEnabled()) {
- LOGGER.error("Job is already being monitored for job: " + jobId);
+ LOGGER.error("Job {} is already being monitored", jobId);
}
return;
}
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "monitoring started for job id: " + jobId);
+ LOGGER.log(level, "Monitoring started for job {}", jobId);
}
} else {
if (LOGGER.isEnabled(level)) {
- LOGGER.info("No listener was found for the entity: " + entityId);
+ LOGGER.info("No listener was found for the entity {} for job {}", entityId, jobId);
}
}
jobId2EntityId.put(jobId, entityId);
@@ -146,14 +146,14 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
public synchronized void notifyJobFinish(JobId jobId, JobStatus jobStatus, List<Exception> exceptions)
throws HyracksException {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Getting notified of job finish for JobId: " + jobId);
+ LOGGER.log(level, "Getting notified of job finish for job {}", jobId);
}
EntityId entityId = jobId2EntityId.get(jobId);
if (entityId != null) {
add(new ActiveEvent(jobId, Kind.JOB_FINISHED, entityId, Pair.of(jobStatus, exceptions)));
} else {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "no need to notify job finish");
+ LOGGER.log(level, "No need to notify JOB_FINISHED for job {}", jobId);
}
}
}
@@ -169,11 +169,11 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
@Override
public IActiveEntityEventsListener getListener(EntityId entityId) {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "getActiveEntityListener(EntityId entityId) was called with entity " + entityId);
+ LOGGER.log(level, "getActiveEntityListener was called with entity {}", entityId);
}
IActiveEntityEventsListener listener = entityEventListeners.get(entityId);
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Listener found: " + listener);
+ LOGGER.log(level, "Listener found: {}", listener);
}
return entityEventListeners.get(entityId);
}
@@ -192,8 +192,7 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
throw new RuntimeDataException(ErrorCode.ACTIVE_NOTIFICATION_HANDLER_IS_SUSPENDED);
}
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "registerListener(IActiveEntityEventsListener listener) was called for the entity "
- + listener.getEntityId());
+ LOGGER.log(level, "registerListener was called for the entity {}", listener.getEntityId());
}
if (entityEventListeners.containsKey(listener.getEntityId())) {
throw new RuntimeDataException(ErrorCode.ACTIVE_ENTITY_IS_ALREADY_REGISTERED, listener.getEntityId());
@@ -207,8 +206,7 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
throw new RuntimeDataException(ErrorCode.ACTIVE_NOTIFICATION_HANDLER_IS_SUSPENDED);
}
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "unregisterListener(IActiveEntityEventsListener listener) was called for the entity "
- + listener.getEntityId());
+ LOGGER.log(level, "unregisterListener was called for the entity {}", listener.getEntityId());
}
IActiveEntityEventsListener registeredListener = entityEventListeners.remove(listener.getEntityId());
if (registeredListener == null) {
@@ -226,7 +224,7 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
for (IActiveEntityEventsListener listener : getEventListeners()) {
synchronized (listener) {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Entity " + listener.getEntityId() + " is " + listener.getState());
+ LOGGER.log(level, "Entity {} is {}", listener.getEntityId(), listener.getState());
}
listener.notifyAll();
}
@@ -276,11 +274,11 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
public void resumeOrHalt(IActiveEntityEventsListener listener, MetadataProvider metadataProvider) {
try {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Resuming " + listener.getEntityId());
+ LOGGER.log(level, "Resuming {}", listener.getEntityId());
}
((ActiveEntityEventsListener) listener).resume(metadataProvider);
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, listener.getEntityId() + " resumed");
+ LOGGER.log(level, "{} resumed", listener.getEntityId());
}
} catch (Throwable th) { // NOSONAR must halt in case of any failure
LOGGER.error("Resume active failed", th);
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryRequest.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryRequest.java
index 6154faa890..65d1039aff 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryRequest.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryRequest.java
@@ -50,6 +50,7 @@ public class CancelQueryRequest implements ICcAddressedMessage {
final IRequestTracker requestTracker = appCtx.getRequestTracker();
IClientRequest req = uuid != null ? requestTracker.get(uuid) : requestTracker.getByClientContextId(contextId);
RequestStatus status;
+ String requestId = "";
if (req == null) {
LOGGER.log(Level.INFO, "No request found for uuid {} or context id {}", uuid, contextId);
@@ -59,7 +60,8 @@ public class CancelQueryRequest implements ICcAddressedMessage {
status = RequestStatus.REJECTED;
} else {
try {
- requestTracker.cancel(req.getId());
+ requestId = req.getId();
+ requestTracker.cancel(requestId);
status = RequestStatus.SUCCESS;
} catch (Exception e) {
LOGGER.log(Level.WARN, "unexpected exception thrown from cancel", e);
@@ -67,6 +69,10 @@ public class CancelQueryRequest implements ICcAddressedMessage {
}
}
}
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Sending CancelQueryResponse to {}. requestId:{}, uuid:{}, contextId:{}, status:{}", nodeId,
+ requestId, uuid, contextId, status);
+ }
CancelQueryResponse response = new CancelQueryResponse(reqId, status);
CCMessageBroker messageBroker = (CCMessageBroker) appCtx.getServiceContext().getMessageBroker();
try {
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
index 0a5e0330a8..9b722df12c 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
@@ -243,7 +243,7 @@ public class ExecuteStatementRequestMessage implements ICcAddressedMessage {
@Override
public String toString() {
- return String.format("%s(id=%s, from=%s): %s", getClass().getSimpleName(), requestMessageId, requestNodeId,
- LogRedactionUtil.statement(statementsText));
+ return String.format("%s(id=%s, from=%s, uuid=%s): %s", getClass().getSimpleName(), requestMessageId,
+ requestNodeId, LogRedactionUtil.statement(statementsText), requestReference.getUuid());
}
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index e098a89365..afdd0359a4 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@ -4837,6 +4837,10 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
// ensure request not cancelled before running job
ensureNotCancelled(clientRequest);
final JobId jobId = JobUtils.runJob(hcc, jobSpec, jobFlags, false);
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("createAndRunJob jobId:{}, uuid:{}", jobId,
+ requestParameters.getRequestReference().getUuid());
+ }
clientRequest.setJobId(jobId);
if (jId != null) {
jId.setValue(jobId);
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/NetworkAddress.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/NetworkAddress.java
index d74f50044c..75fbb923f3 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/NetworkAddress.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/NetworkAddress.java
@@ -25,6 +25,7 @@ import java.io.Serializable;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
+import java.util.Objects;
import org.apache.hyracks.api.io.IWritable;
@@ -82,6 +83,10 @@ public final class NetworkAddress implements IWritable, Serializable {
return inetSocketAddress;
}
+ public InetSocketAddress toInetSocketAddress() {
+ return new InetSocketAddress(address, port);
+ }
+
public int getPort() {
return port;
}
@@ -102,7 +107,7 @@ public final class NetworkAddress implements IWritable, Serializable {
return false;
}
NetworkAddress on = (NetworkAddress) o;
- return on.port == port && on.address == address;
+ return on.port == port && Objects.equals(on.address, address);
}
@Override
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/CancelJobWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/CancelJobWork.java
index e3135df076..9a08e8e2d6 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/CancelJobWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/CancelJobWork.java
@@ -47,4 +47,9 @@ public class CancelJobWork extends SynchronizableWork {
callback.setException(e);
}
}
+
+ @Override
+ public String toString() {
+ return getName() + " jobId:" + jobId;
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/JobletCleanupNotificationWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/JobletCleanupNotificationWork.java
index 727793b219..c3a09f9da3 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/JobletCleanupNotificationWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/JobletCleanupNotificationWork.java
@@ -37,7 +37,7 @@ import org.apache.logging.log4j.Logger;
public class JobletCleanupNotificationWork extends AbstractHeartbeatWork {
private static final Logger LOGGER = LogManager.getLogger();
- private JobId jobId;
+ private final JobId jobId;
public JobletCleanupNotificationWork(ClusterControllerService ccs, JobId jobId, String nodeId) {
super(ccs, nodeId, null);
@@ -77,4 +77,9 @@ public class JobletCleanupNotificationWork extends AbstractHeartbeatWork {
}
}
}
+
+ @Override
+ public String toString() {
+ return getName() + " jobId:" + jobId + ", nodeId:" + nodeId;
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/RegisterNodeWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/RegisterNodeWork.java
index b1700febb5..ec21785baa 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/RegisterNodeWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/RegisterNodeWork.java
@@ -18,6 +18,7 @@
*/
package org.apache.hyracks.control.cc.work;
+import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
@@ -50,8 +51,9 @@ public class RegisterNodeWork extends SynchronizableWork {
protected void doRun() throws Exception {
String id = reg.getNodeId();
LOGGER.info("registering node: {}", id);
- NodeControllerRemoteProxy nc = new NodeControllerRemoteProxy(ccs.getCcId(),
- ccs.getClusterIPC().getReconnectingHandle(reg.getNodeControllerAddress().resolveInetSocketAddress()));
+ InetSocketAddress ncAddress = reg.getNodeControllerAddress().toInetSocketAddress();
+ NodeControllerRemoteProxy nc =
+ new NodeControllerRemoteProxy(ccs.getCcId(), ccs.getClusterIPC().getReconnectingHandle(ncAddress));
INodeManager nodeManager = ccs.getNodeManager();
NodeParameters params = new NodeParameters();
params.setClusterControllerInfo(ccs.getClusterControllerInfo());
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/WaitForJobCompletionWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/WaitForJobCompletionWork.java
index 14d92fb6f5..ed3e574323 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/WaitForJobCompletionWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/WaitForJobCompletionWork.java
@@ -87,4 +87,9 @@ public class WaitForJobCompletionWork extends SynchronizableWork {
});
}
}
+
+ @Override
+ public String toString() {
+ return getName() + " jobId:" + jobId;
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
index 8c9cbfb6c4..f69d106f1d 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
@@ -426,7 +426,7 @@ public class NodeControllerService implements IControllerService {
NodeParameters nodeParameters = ccc.getNodeParameters();
// Start heartbeat generator.
heartbeatManagers.computeIfAbsent(ccId, newCcId -> HeartbeatManager.init(this, ccc, hbTask.getHeartbeatData(),
- nodeRegistration.getNodeControllerAddress().resolveInetSocketAddress()));
+ nodeRegistration.getNodeControllerAddress().toInetSocketAddress()));
if (!ccTimers.containsKey(ccId) && nodeParameters.getProfileDumpPeriod() > 0) {
Timer ccTimer = new Timer("Timer-" + ccId, true);
// Schedule profile dump generator.
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/AbortTasksWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/AbortTasksWork.java
index f47e1ce957..bfe370659f 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/AbortTasksWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/AbortTasksWork.java
@@ -70,4 +70,9 @@ public class AbortTasksWork extends AbstractWork {
"Joblet couldn't be found. Tasks of job " + jobId + " have all either completed or failed");
}
}
+
+ @Override
+ public String toString() {
+ return getName() + " jobId:" + jobId;
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/CleanupJobletWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/CleanupJobletWork.java
index ae2cfa098e..75edd38b00 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/CleanupJobletWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/CleanupJobletWork.java
@@ -35,7 +35,7 @@ public class CleanupJobletWork extends AbstractWork {
private final JobId jobId;
- private JobStatus status;
+ private final JobStatus status;
public CleanupJobletWork(NodeControllerService ncs, JobId jobId, JobStatus status) {
this.ncs = ncs;
@@ -54,4 +54,9 @@ public class CleanupJobletWork extends AbstractWork {
joblet.cleanup(status);
}
}
+
+ @Override
+ public String toString() {
+ return getName() + " jobId:" + jobId + ", status:" + status;
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/StartTasksWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/StartTasksWork.java
index f6c144db93..f277046496 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/StartTasksWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/StartTasksWork.java
@@ -315,4 +315,9 @@ public class StartTasksWork extends AbstractWork {
}
return channelsForInputConnectors;
}
+
+ @Override
+ public String toString() {
+ return getName() + " jobId:" + jobId;
+ }
}
[asterixdb] 12/20: Merge branch 'gerrit/stabilization-5949a1cb71' into 'gerrit/mad-hatter'
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 84c47cf88a3c22b8f5c224cc38dd3c86c886abfe
Merge: 85a3f577df cf40ec4341
Author: Michael Blow <mi...@couchbase.com>
AuthorDate: Fri Jul 14 18:30:18 2023 -0400
Merge branch 'gerrit/stabilization-5949a1cb71' into 'gerrit/mad-hatter'
Change-Id: Id0866a023e2c67f6a6d33717f085bed9b3ee9915
...rcontent.com_netty_netty_netty-4.1.94.Final_NOTICE.txt} | 0
hyracks-fullstack/pom.xml | 14 +++++++-------
2 files changed, 7 insertions(+), 7 deletions(-)
diff --cc hyracks-fullstack/pom.xml
index 5e07d9be9d,f3f75532e2..8311c5cfad
--- a/hyracks-fullstack/pom.xml
+++ b/hyracks-fullstack/pom.xml
@@@ -112,13 -111,8 +112,13 @@@
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-handler</artifactId>
- <version>4.1.87.Final</version>
+ <version>4.1.94.Final</version>
</dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-transport-classes-epoll</artifactId>
- <version>4.1.87.Final</version>
++ <version>4.1.94.Final</version>
+ </dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
[asterixdb] 01/20: [NO ISSUE][HYR][HTTP] Prevent Netty buffer leaks on close() errors
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 677f1ae6975d5ddd9fc509cc689fa18281fd6091
Author: Michael Blow <mb...@apache.org>
AuthorDate: Thu Jun 29 18:34:59 2023 -0400
[NO ISSUE][HYR][HTTP] Prevent Netty buffer leaks on close() errors
Change-Id: Ibe5836eb61b3f01b8d95820f20c55269e98a3118
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17624
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Michael Blow <mb...@apache.org>
Reviewed-by: Murtadha Al Hubail <mh...@apache.org>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
---
.../org/apache/hyracks/api/util/InvokeUtil.java | 40 ++++++++++++++-
.../hyracks/http/server/ChunkedResponse.java | 57 ++++++++++++++--------
.../org/apache/hyracks/util/IOThrowingAction.java | 33 +++++++++++++
3 files changed, 108 insertions(+), 22 deletions(-)
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/InvokeUtil.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/InvokeUtil.java
index 7d04cf2254..57f9750459 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/InvokeUtil.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/InvokeUtil.java
@@ -29,6 +29,7 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.util.ComputingAction;
import org.apache.hyracks.util.IDelay;
import org.apache.hyracks.util.IOInterruptibleAction;
+import org.apache.hyracks.util.IOThrowingAction;
import org.apache.hyracks.util.IRetryPolicy;
import org.apache.hyracks.util.InterruptibleAction;
import org.apache.hyracks.util.Span;
@@ -188,7 +189,7 @@ public class InvokeUtil {
}
}
- @SuppressWarnings({ "squid:S1181", "squid:S1193" }) // catching Throwable, instanceof of exception
+ @SuppressWarnings({ "squid:S1181", "squid:S1193", "ConstantConditions" }) // catching Throwable, instanceofs
public static void tryWithCleanups(ThrowingAction action, ThrowingAction... cleanups) throws Exception {
Throwable savedT = null;
boolean suppressedInterrupted = false;
@@ -225,6 +226,43 @@ public class InvokeUtil {
}
}
+ @SuppressWarnings({ "squid:S1181", "squid:S1193", "ConstantConditions" }) // catching Throwable, instanceofs
+ public static void tryIoWithCleanups(IOThrowingAction action, IOThrowingAction... cleanups) throws IOException {
+ Throwable savedT = null;
+ boolean suppressedInterrupted = false;
+ try {
+ action.run();
+ } catch (Throwable t) {
+ savedT = t;
+ } finally {
+ for (IOThrowingAction cleanup : cleanups) {
+ try {
+ cleanup.run();
+ } catch (Throwable t) {
+ if (savedT != null) {
+ savedT.addSuppressed(t);
+ suppressedInterrupted = suppressedInterrupted || t instanceof InterruptedException;
+ } else {
+ savedT = t;
+ }
+ }
+ }
+ }
+ if (savedT == null) {
+ return;
+ }
+ if (suppressedInterrupted) {
+ Thread.currentThread().interrupt();
+ }
+ if (savedT instanceof Error) {
+ throw (Error) savedT;
+ } else if (savedT instanceof IOException) {
+ throw (IOException) savedT;
+ } else {
+ throw HyracksDataException.create(savedT);
+ }
+ }
+
/**
* Runs the supplied action, after suspending any pending interruption. An error will be logged if
* the action is itself interrupted.
diff --git a/hyracks-fullstack/hyracks/hyracks-http/src/main/java/org/apache/hyracks/http/server/ChunkedResponse.java b/hyracks-fullstack/hyracks/hyracks-http/src/main/java/org/apache/hyracks/http/server/ChunkedResponse.java
index e00c51903e..740af2fe34 100644
--- a/hyracks-fullstack/hyracks/hyracks-http/src/main/java/org/apache/hyracks/http/server/ChunkedResponse.java
+++ b/hyracks-fullstack/hyracks/hyracks-http/src/main/java/org/apache/hyracks/http/server/ChunkedResponse.java
@@ -25,6 +25,8 @@ import java.io.PrintWriter;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.util.InvokeUtil;
import org.apache.hyracks.http.api.IServletResponse;
import org.apache.hyracks.http.server.utils.HttpUtil;
import org.apache.logging.log4j.Level;
@@ -45,6 +47,7 @@ import io.netty.handler.codec.http.HttpObject;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.util.ReferenceCountUtil;
/**
* A chunked http response. Here is how it is expected to work:
@@ -114,28 +117,40 @@ public class ChunkedResponse implements IServletResponse {
@Override
public void close() throws IOException {
- if (writer != null) {
- writer.close();
- } else {
- outputStream.close();
- }
- if (errorBuf == null && response.status() == HttpResponseStatus.OK) {
- if (!done) {
- respond(LastHttpContent.EMPTY_LAST_CONTENT);
- }
- } else {
- // There was an error
- if (headerSent) {
- LOGGER.log(Level.WARN, "Error after header write of chunked response");
- if (errorBuf != null) {
- errorBuf.release();
+ try {
+ InvokeUtil.tryIoWithCleanups(() -> {
+ if (writer != null) {
+ writer.close();
+ } else {
+ outputStream.close();
+ }
+ if (errorBuf == null && response.status() == HttpResponseStatus.OK) {
+ if (!done) {
+ respond(LastHttpContent.EMPTY_LAST_CONTENT);
+ }
+ } else {
+ // There was an error
+ if (headerSent) {
+ LOGGER.log(Level.WARN, "Error after header write of chunked response");
+ future = ctx.channel().close().addListener(handler);
+ } else {
+ // we didn't send anything to the user, we need to send an non-chunked error response
+ fullResponse(response.protocolVersion(), response.status(),
+ errorBuf == null ? ctx.alloc().buffer(0, 0) : errorBuf, response.headers());
+ // The responsibility of releasing the error buffer is now with the netty pipeline since it is
+ // forwarded within the http content. We must nullify buffer to avoid releasing the buffer twice.
+ errorBuf = null;
+ }
}
- future = ctx.channel().close().addListener(handler);
- } else {
- // we didn't send anything to the user, we need to send an non-chunked error response
- fullResponse(response.protocolVersion(), response.status(),
- errorBuf == null ? ctx.alloc().buffer(0, 0) : errorBuf, response.headers());
- }
+ }, outputStream::close, () -> {
+ ReferenceCountUtil.release(errorBuf);
+ // We must nullify buffer to avoid releasing the buffer twice in case of duplicate close()
+ errorBuf = null;
+ });
+ } catch (IOException e) {
+ throw e;
+ } catch (Exception e) {
+ throw HyracksDataException.create(e);
}
done = true;
}
diff --git a/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/IOThrowingAction.java b/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/IOThrowingAction.java
new file mode 100644
index 0000000000..b9430d84dd
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/IOThrowingAction.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hyracks.util;
+
+import java.io.IOException;
+
+@FunctionalInterface
+public interface IOThrowingAction {
+ void run() throws IOException; // NOSONAR
+
+ static ComputingAction<Void> asComputingAction(IOThrowingAction action) {
+ return () -> {
+ action.run();
+ return null;
+ };
+ }
+}
[asterixdb] 06/20: [ASTERIXDB-3216][NET] Always resolve NC address on node registration
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 0dded85deff3112d1562741d1e277d47faad0d72
Author: Ali Alsuliman <al...@gmail.com>
AuthorDate: Tue Jul 4 21:30:31 2023 -0700
[ASTERIXDB-3216][NET] Always resolve NC address on node registration
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
Always resolve NC address on node registration.
- add some loggings
(cherry picked from neo commit 1e367fd1)
Change-Id: I37477de0a932439e301d8ff6f88de1355d612736
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17568
Reviewed-by: Michael Blow <mb...@apache.org>
Reviewed-by: Ali Alsuliman <al...@gmail.com>
Tested-by: Michael Blow <mb...@apache.org>
---
.../http/server/NCQueryCancellationServlet.java | 3 ++
.../api/http/server/NCQueryServiceServlet.java | 3 +-
.../app/active/ActiveEntityEventsListener.java | 10 ++---
.../app/active/ActiveNotificationHandler.java | 52 +++++++++++-----------
.../asterix/app/message/CancelQueryRequest.java | 8 +++-
.../message/ExecuteStatementRequestMessage.java | 4 +-
.../asterix/app/translator/QueryTranslator.java | 4 ++
.../apache/hyracks/api/comm/NetworkAddress.java | 7 ++-
.../hyracks/control/cc/work/CancelJobWork.java | 5 +++
.../cc/work/JobletCleanupNotificationWork.java | 7 ++-
.../hyracks/control/cc/work/RegisterNodeWork.java | 6 ++-
.../control/cc/work/WaitForJobCompletionWork.java | 5 +++
.../hyracks/control/nc/NodeControllerService.java | 2 +-
.../hyracks/control/nc/work/AbortTasksWork.java | 5 +++
.../hyracks/control/nc/work/CleanupJobletWork.java | 7 ++-
.../hyracks/control/nc/work/StartTasksWork.java | 5 +++
16 files changed, 91 insertions(+), 42 deletions(-)
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryCancellationServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryCancellationServlet.java
index 41f2a53dc2..b2134dcfc6 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryCancellationServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryCancellationServlet.java
@@ -67,6 +67,9 @@ public class NCQueryCancellationServlet extends AbstractServlet {
CancelQueryRequest cancelQueryMessage = new CancelQueryRequest(serviceCtx.getNodeId(),
cancelQueryFuture.getFutureId(), uuid, clientContextId);
// TODO(mblow): multicc -- need to send cancellation to the correct cc
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Sending CancelQueryRequest with uuid:{}, clientContextID:{}", uuid, clientContextId);
+ }
messageBroker.sendMessageToPrimaryCC(cancelQueryMessage);
CancelQueryResponse cancelResponse =
(CancelQueryResponse) cancelQueryFuture.get(DEFAULT_NC_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryServiceServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryServiceServlet.java
index 1ec74055d0..9fa479c398 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryServiceServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryServiceServlet.java
@@ -165,7 +165,8 @@ public class NCQueryServiceServlet extends QueryServiceServlet {
CancelQueryRequest cancelQueryMessage =
new CancelQueryRequest(nodeId, cancelQueryFuture.getFutureId(), uuid, clientContextID);
// TODO(mblow): multicc -- need to send cancellation to the correct cc
- LOGGER.info("Cancelling query due to {}", exception.getClass().getSimpleName());
+ LOGGER.info("Cancelling query with uuid:{}, clientContextID:{} due to {}", uuid, clientContextID,
+ exception.getClass().getSimpleName());
messageBroker.sendMessageToPrimaryCC(cancelQueryMessage);
if (wait) {
cancelQueryFuture.get(ExecuteStatementRequestMessage.DEFAULT_QUERY_CANCELLATION_WAIT_MILLIS,
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
index ddd3d64682..3fd339e8df 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
@@ -195,12 +195,12 @@ public abstract class ActiveEntityEventsListener implements IActiveEntityControl
@SuppressWarnings("unchecked")
protected void finish(ActiveEvent event) throws HyracksDataException {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "the job {} finished", jobId);
+ LOGGER.log(level, "Active job {} finished", jobId);
}
JobId lastJobId = jobId;
if (numRegistered != numDeRegistered) {
LOGGER.log(Level.WARN,
- "the job {} finished with reported runtime registrations = {} and deregistrations = {}", jobId,
+ "Active job {} finished with reported runtime registrations = {} and deregistrations = {}", jobId,
numRegistered, numDeRegistered);
}
jobId = null;
@@ -208,7 +208,7 @@ public abstract class ActiveEntityEventsListener implements IActiveEntityControl
JobStatus jobStatus = status.getLeft();
List<Exception> exceptions = status.getRight();
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "The job finished with status: {}", jobStatus);
+ LOGGER.log(level, "Active job {} finished with status {}", lastJobId, jobStatus);
}
if (!jobSuccessfullyTerminated(jobStatus)) {
jobFailure = exceptions.isEmpty() ? new RuntimeDataException(ErrorCode.UNREPORTED_TASK_FAILURE_EXCEPTION)
@@ -372,7 +372,7 @@ public abstract class ActiveEntityEventsListener implements IActiveEntityControl
@Override
public synchronized void recover() {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Recover is called on " + entityId);
+ LOGGER.log(level, "Recover is called on {}", entityId);
}
if (retryPolicyFactory == NoRetryPolicyFactory.INSTANCE) {
LOGGER.log(level, "But it has no recovery policy, so it is set to permanent failure");
@@ -456,7 +456,7 @@ public abstract class ActiveEntityEventsListener implements IActiveEntityControl
try {
metadataProvider.getApplicationContext().getHcc().cancelJob(jobId);
} catch (Throwable th) {
- LOGGER.warn("Failed to cancel active job", th);
+ LOGGER.warn("Failed to cancel active job {}", jobId, th);
e.addSuppressed(th);
}
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java
index 284929f5ae..6b3581e66c 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java
@@ -66,22 +66,24 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
@Override
protected void handle(ActiveEvent event) {
- EntityId entityId = jobId2EntityId.get(event.getJobId());
+ JobId jobId = event.getJobId();
+ Kind eventKind = event.getEventKind();
+ EntityId entityId = jobId2EntityId.get(jobId);
if (entityId != null) {
IActiveEntityEventsListener listener = entityEventListeners.get(entityId);
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Next event is of type " + event.getEventKind());
+ LOGGER.log(level, "Next event is {} for job {}", eventKind, jobId);
}
- if (event.getEventKind() == Kind.JOB_FINISHED) {
- LOGGER.log(level, "Removing the job");
- jobId2EntityId.remove(event.getJobId());
+ if (eventKind == Kind.JOB_FINISHED) {
+ LOGGER.log(level, "Removing job {}", jobId);
+ jobId2EntityId.remove(jobId);
}
if (listener != null) {
LOGGER.log(level, "Notifying the listener");
listener.notify(event);
}
} else {
- LOGGER.log(Level.ERROR, "Entity not found for received message for job " + event.getJobId());
+ LOGGER.log(Level.ERROR, "Entity not found for event {} for job {}", eventKind, jobId);
}
}
@@ -90,45 +92,43 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
@Override
public void notifyJobCreation(JobId jobId, JobSpecification jobSpecification) throws HyracksDataException {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level,
- "notifyJobCreation(JobId jobId, JobSpecification jobSpecification) was called with jobId = "
- + jobId);
+ LOGGER.log(level, "notifyJobCreation was called for job {}", jobId);
}
Object property = jobSpecification.getProperty(ACTIVE_ENTITY_PROPERTY_NAME);
if (!(property instanceof EntityId)) {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Job is not of type active job. property found to be: " + property);
+ LOGGER.log(level, "Job {} is not of type active job. property found to be {}", jobId, property);
}
return;
}
EntityId entityId = (EntityId) property;
monitorJob(jobId, entityId);
boolean found = jobId2EntityId.get(jobId) != null;
- LOGGER.log(level, "Job was found to be: " + (found ? "Active" : "Inactive"));
+ LOGGER.log(level, "Job {} was found to be {}", jobId, (found ? "Active" : "Inactive"));
add(new ActiveEvent(jobId, Kind.JOB_CREATED, entityId, jobSpecification));
}
private synchronized void monitorJob(JobId jobId, EntityId entityId) {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "monitorJob(JobId jobId, ActiveJob activeJob) called with job id: " + jobId);
+ LOGGER.log(level, "monitorJob was called for job {}", jobId);
}
boolean found = jobId2EntityId.get(jobId) != null;
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Job was found to be: " + (found ? "Active" : "Inactive"));
+ LOGGER.log(level, "Job {} was found to be {}", jobId, (found ? "Active" : "Inactive"));
}
if (entityEventListeners.containsKey(entityId)) {
if (jobId2EntityId.containsKey(jobId)) {
if (LOGGER.isErrorEnabled()) {
- LOGGER.error("Job is already being monitored for job: " + jobId);
+ LOGGER.error("Job {} is already being monitored", jobId);
}
return;
}
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "monitoring started for job id: " + jobId);
+ LOGGER.log(level, "Monitoring started for job {}", jobId);
}
} else {
if (LOGGER.isEnabled(level)) {
- LOGGER.info("No listener was found for the entity: " + entityId);
+ LOGGER.info("No listener was found for the entity {} for job {}", entityId, jobId);
}
}
jobId2EntityId.put(jobId, entityId);
@@ -146,14 +146,14 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
public synchronized void notifyJobFinish(JobId jobId, JobStatus jobStatus, List<Exception> exceptions)
throws HyracksException {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Getting notified of job finish for JobId: " + jobId);
+ LOGGER.log(level, "Getting notified of job finish for job {}", jobId);
}
EntityId entityId = jobId2EntityId.get(jobId);
if (entityId != null) {
add(new ActiveEvent(jobId, Kind.JOB_FINISHED, entityId, Pair.of(jobStatus, exceptions)));
} else {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "no need to notify job finish");
+ LOGGER.log(level, "No need to notify JOB_FINISHED for job {}", jobId);
}
}
}
@@ -169,11 +169,11 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
@Override
public IActiveEntityEventsListener getListener(EntityId entityId) {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "getActiveEntityListener(EntityId entityId) was called with entity " + entityId);
+ LOGGER.log(level, "getActiveEntityListener was called with entity {}", entityId);
}
IActiveEntityEventsListener listener = entityEventListeners.get(entityId);
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Listener found: " + listener);
+ LOGGER.log(level, "Listener found: {}", listener);
}
return entityEventListeners.get(entityId);
}
@@ -192,8 +192,7 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
throw new RuntimeDataException(ErrorCode.ACTIVE_NOTIFICATION_HANDLER_IS_SUSPENDED);
}
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "registerListener(IActiveEntityEventsListener listener) was called for the entity "
- + listener.getEntityId());
+ LOGGER.log(level, "registerListener was called for the entity {}", listener.getEntityId());
}
if (entityEventListeners.containsKey(listener.getEntityId())) {
throw new RuntimeDataException(ErrorCode.ACTIVE_ENTITY_IS_ALREADY_REGISTERED, listener.getEntityId());
@@ -207,8 +206,7 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
throw new RuntimeDataException(ErrorCode.ACTIVE_NOTIFICATION_HANDLER_IS_SUSPENDED);
}
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "unregisterListener(IActiveEntityEventsListener listener) was called for the entity "
- + listener.getEntityId());
+ LOGGER.log(level, "unregisterListener was called for the entity {}", listener.getEntityId());
}
IActiveEntityEventsListener registeredListener = entityEventListeners.remove(listener.getEntityId());
if (registeredListener == null) {
@@ -226,7 +224,7 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
for (IActiveEntityEventsListener listener : getEventListeners()) {
synchronized (listener) {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Entity " + listener.getEntityId() + " is " + listener.getState());
+ LOGGER.log(level, "Entity {} is {}", listener.getEntityId(), listener.getState());
}
listener.notifyAll();
}
@@ -276,11 +274,11 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
public void resumeOrHalt(IActiveEntityEventsListener listener, MetadataProvider metadataProvider) {
try {
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, "Resuming " + listener.getEntityId());
+ LOGGER.log(level, "Resuming {}", listener.getEntityId());
}
((ActiveEntityEventsListener) listener).resume(metadataProvider);
if (LOGGER.isEnabled(level)) {
- LOGGER.log(level, listener.getEntityId() + " resumed");
+ LOGGER.log(level, "{} resumed", listener.getEntityId());
}
} catch (Throwable th) { // NOSONAR must halt in case of any failure
LOGGER.error("Resume active failed", th);
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryRequest.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryRequest.java
index 6154faa890..65d1039aff 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryRequest.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryRequest.java
@@ -50,6 +50,7 @@ public class CancelQueryRequest implements ICcAddressedMessage {
final IRequestTracker requestTracker = appCtx.getRequestTracker();
IClientRequest req = uuid != null ? requestTracker.get(uuid) : requestTracker.getByClientContextId(contextId);
RequestStatus status;
+ String requestId = "";
if (req == null) {
LOGGER.log(Level.INFO, "No request found for uuid {} or context id {}", uuid, contextId);
@@ -59,7 +60,8 @@ public class CancelQueryRequest implements ICcAddressedMessage {
status = RequestStatus.REJECTED;
} else {
try {
- requestTracker.cancel(req.getId());
+ requestId = req.getId();
+ requestTracker.cancel(requestId);
status = RequestStatus.SUCCESS;
} catch (Exception e) {
LOGGER.log(Level.WARN, "unexpected exception thrown from cancel", e);
@@ -67,6 +69,10 @@ public class CancelQueryRequest implements ICcAddressedMessage {
}
}
}
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Sending CancelQueryResponse to {}. requestId:{}, uuid:{}, contextId:{}, status:{}", nodeId,
+ requestId, uuid, contextId, status);
+ }
CancelQueryResponse response = new CancelQueryResponse(reqId, status);
CCMessageBroker messageBroker = (CCMessageBroker) appCtx.getServiceContext().getMessageBroker();
try {
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
index 0a5e0330a8..9b722df12c 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
@@ -243,7 +243,7 @@ public class ExecuteStatementRequestMessage implements ICcAddressedMessage {
@Override
public String toString() {
- return String.format("%s(id=%s, from=%s): %s", getClass().getSimpleName(), requestMessageId, requestNodeId,
- LogRedactionUtil.statement(statementsText));
+ return String.format("%s(id=%s, from=%s, uuid=%s): %s", getClass().getSimpleName(), requestMessageId,
+ requestNodeId, LogRedactionUtil.statement(statementsText), requestReference.getUuid());
}
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index 298468c3fb..df7d8d64e0 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@ -4367,6 +4367,10 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
// ensure request not cancelled before running job
ensureNotCancelled(clientRequest);
final JobId jobId = JobUtils.runJob(hcc, jobSpec, jobFlags, false);
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("createAndRunJob jobId:{}, uuid:{}", jobId,
+ requestParameters.getRequestReference().getUuid());
+ }
clientRequest.setJobId(jobId);
if (jId != null) {
jId.setValue(jobId);
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/NetworkAddress.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/NetworkAddress.java
index d74f50044c..75fbb923f3 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/NetworkAddress.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/NetworkAddress.java
@@ -25,6 +25,7 @@ import java.io.Serializable;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
+import java.util.Objects;
import org.apache.hyracks.api.io.IWritable;
@@ -82,6 +83,10 @@ public final class NetworkAddress implements IWritable, Serializable {
return inetSocketAddress;
}
+ public InetSocketAddress toInetSocketAddress() {
+ return new InetSocketAddress(address, port);
+ }
+
public int getPort() {
return port;
}
@@ -102,7 +107,7 @@ public final class NetworkAddress implements IWritable, Serializable {
return false;
}
NetworkAddress on = (NetworkAddress) o;
- return on.port == port && on.address == address;
+ return on.port == port && Objects.equals(on.address, address);
}
@Override
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/CancelJobWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/CancelJobWork.java
index e3135df076..9a08e8e2d6 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/CancelJobWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/CancelJobWork.java
@@ -47,4 +47,9 @@ public class CancelJobWork extends SynchronizableWork {
callback.setException(e);
}
}
+
+ @Override
+ public String toString() {
+ return getName() + " jobId:" + jobId;
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/JobletCleanupNotificationWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/JobletCleanupNotificationWork.java
index 727793b219..c3a09f9da3 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/JobletCleanupNotificationWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/JobletCleanupNotificationWork.java
@@ -37,7 +37,7 @@ import org.apache.logging.log4j.Logger;
public class JobletCleanupNotificationWork extends AbstractHeartbeatWork {
private static final Logger LOGGER = LogManager.getLogger();
- private JobId jobId;
+ private final JobId jobId;
public JobletCleanupNotificationWork(ClusterControllerService ccs, JobId jobId, String nodeId) {
super(ccs, nodeId, null);
@@ -77,4 +77,9 @@ public class JobletCleanupNotificationWork extends AbstractHeartbeatWork {
}
}
}
+
+ @Override
+ public String toString() {
+ return getName() + " jobId:" + jobId + ", nodeId:" + nodeId;
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/RegisterNodeWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/RegisterNodeWork.java
index b1700febb5..ec21785baa 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/RegisterNodeWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/RegisterNodeWork.java
@@ -18,6 +18,7 @@
*/
package org.apache.hyracks.control.cc.work;
+import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
@@ -50,8 +51,9 @@ public class RegisterNodeWork extends SynchronizableWork {
protected void doRun() throws Exception {
String id = reg.getNodeId();
LOGGER.info("registering node: {}", id);
- NodeControllerRemoteProxy nc = new NodeControllerRemoteProxy(ccs.getCcId(),
- ccs.getClusterIPC().getReconnectingHandle(reg.getNodeControllerAddress().resolveInetSocketAddress()));
+ InetSocketAddress ncAddress = reg.getNodeControllerAddress().toInetSocketAddress();
+ NodeControllerRemoteProxy nc =
+ new NodeControllerRemoteProxy(ccs.getCcId(), ccs.getClusterIPC().getReconnectingHandle(ncAddress));
INodeManager nodeManager = ccs.getNodeManager();
NodeParameters params = new NodeParameters();
params.setClusterControllerInfo(ccs.getClusterControllerInfo());
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/WaitForJobCompletionWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/WaitForJobCompletionWork.java
index 911bedf767..4ffe9a09d6 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/WaitForJobCompletionWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/WaitForJobCompletionWork.java
@@ -84,4 +84,9 @@ public class WaitForJobCompletionWork extends SynchronizableWork {
});
}
}
+
+ @Override
+ public String toString() {
+ return getName() + " jobId:" + jobId;
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
index 8c9cbfb6c4..f69d106f1d 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
@@ -426,7 +426,7 @@ public class NodeControllerService implements IControllerService {
NodeParameters nodeParameters = ccc.getNodeParameters();
// Start heartbeat generator.
heartbeatManagers.computeIfAbsent(ccId, newCcId -> HeartbeatManager.init(this, ccc, hbTask.getHeartbeatData(),
- nodeRegistration.getNodeControllerAddress().resolveInetSocketAddress()));
+ nodeRegistration.getNodeControllerAddress().toInetSocketAddress()));
if (!ccTimers.containsKey(ccId) && nodeParameters.getProfileDumpPeriod() > 0) {
Timer ccTimer = new Timer("Timer-" + ccId, true);
// Schedule profile dump generator.
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/AbortTasksWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/AbortTasksWork.java
index f47e1ce957..bfe370659f 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/AbortTasksWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/AbortTasksWork.java
@@ -70,4 +70,9 @@ public class AbortTasksWork extends AbstractWork {
"Joblet couldn't be found. Tasks of job " + jobId + " have all either completed or failed");
}
}
+
+ @Override
+ public String toString() {
+ return getName() + " jobId:" + jobId;
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/CleanupJobletWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/CleanupJobletWork.java
index ae2cfa098e..75edd38b00 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/CleanupJobletWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/CleanupJobletWork.java
@@ -35,7 +35,7 @@ public class CleanupJobletWork extends AbstractWork {
private final JobId jobId;
- private JobStatus status;
+ private final JobStatus status;
public CleanupJobletWork(NodeControllerService ncs, JobId jobId, JobStatus status) {
this.ncs = ncs;
@@ -54,4 +54,9 @@ public class CleanupJobletWork extends AbstractWork {
joblet.cleanup(status);
}
}
+
+ @Override
+ public String toString() {
+ return getName() + " jobId:" + jobId + ", status:" + status;
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/StartTasksWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/StartTasksWork.java
index 7d84fe9952..442f0d0003 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/StartTasksWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/StartTasksWork.java
@@ -314,4 +314,9 @@ public class StartTasksWork extends AbstractWork {
}
return channelsForInputConnectors;
}
+
+ @Override
+ public String toString() {
+ return getName() + " jobId:" + jobId;
+ }
}
[asterixdb] 19/20: Merge branch 'gerrit/neo' into 'gerrit/trinity'
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 0f3fb3b0b639f58544f4b7209056e0ff376addfa
Merge: f5fe409130 47e2613e0c
Author: Michael Blow <mi...@couchbase.com>
AuthorDate: Fri Jul 28 12:25:11 2023 -0400
Merge branch 'gerrit/neo' into 'gerrit/trinity'
Change-Id: I0b56219450da9db28ef21cd5266dbf1e440424c7
.../http/server/NCQueryCancellationServlet.java | 3 +
.../api/http/server/NCQueryServiceServlet.java | 3 +-
.../app/active/ActiveEntityEventsListener.java | 10 +-
.../app/active/ActiveNotificationHandler.java | 52 +++++-----
.../asterix/app/message/CancelQueryRequest.java | 8 +-
.../message/ExecuteStatementRequestMessage.java | 4 +-
.../org/apache/asterix/app/nc/HaltCallback.java | 4 +-
.../apache/asterix/app/nc/NCAppRuntimeContext.java | 16 +---
.../asterix/app/translator/QueryTranslator.java | 4 +
.../rebalance/IDatasetRebalanceCallback.java | 8 +-
.../rebalance/NoOpDatasetRebalanceCallback.java | 3 +-
.../org/apache/asterix/utils/RebalanceUtil.java | 42 +++++---
.../resources/{cc.conf => cc-cloud-storage.conf} | 3 +-
asterixdb/asterix-app/src/main/resources/cc.conf | 2 +-
asterixdb/asterix-app/src/main/resources/cc2.conf | 2 +-
asterixdb/asterix-app/src/main/resources/cc3.conf | 2 +-
asterixdb/asterix-app/src/main/resources/cc4.conf | 2 +-
.../asterix-app/src/main/resources/cc_no_cbo.conf | 2 +-
.../{cc.conf => cc_static_partitioning.conf} | 3 +-
.../asterix/test/common/ResultExtractor.java | 4 +-
.../test/resources/{cc.conf => cc-analyze.conf} | 11 +--
.../src/test/resources/cc-batch-lookup.conf | 3 +-
.../resources/{cc.conf => cc-cloud-storage.conf} | 3 +-
.../src/test/resources/cc-compression.conf | 2 +-
.../src/test/resources/cc-multipart.conf | 2 +-
...{cc-batch-lookup.conf => cc-single-lookup.conf} | 5 +-
.../asterix-app/src/test/resources/cc-single.conf | 2 +-
.../test/resources/cc-small-txn-log-partition.conf | 2 +-
.../asterix-app/src/test/resources/cc-ssl.conf | 4 +-
.../asterix-app/src/test/resources/cc-storage.conf | 2 +-
asterixdb/asterix-app/src/test/resources/cc.conf | 2 +-
.../api/cluster_state_1/cluster_state_1.1.regexadm | 1 +
.../cluster_state_1_full.1.regexadm | 1 +
.../cluster_state_1_less.1.regexadm | 1 +
.../common/api/IDatasetLifecycleManager.java | 6 +-
.../asterix/common/config/StorageProperties.java | 1 +
.../common/context/BaseOperationTracker.java | 8 +-
.../apache/asterix/common/context/DatasetInfo.java | 31 +++++-
.../common/context/DatasetLifecycleManager.java | 4 +-
.../context/PrimaryIndexOperationTracker.java | 11 +--
.../ioopcallbacks/LSMIOOperationCallback.java | 6 +-
.../ioopcallbacks/LSMIOOperationCallbackTest.java | 7 ++
.../apache/asterix/metadata/RMIClientFactory.java | 13 ++-
.../apache/asterix/metadata/RMIServerFactory.java | 25 ++++-
.../metadata/bootstrap/AsterixStateProxy.java | 15 +--
.../asterix/metadata/entities/NodeGroup.java | 3 +-
.../apache/asterix/metadata/utils/DatasetUtil.java | 5 +-
.../replication/api/ReplicationDestination.java | 32 +++++++
.../management/IndexReplicationManager.java | 99 +++++++++----------
.../management/ReplicationOperation.java | 106 +++++++++++++++++++++
.../replication/sync/ReplicaSynchronizer.java | 2 +-
.../SecondaryIndexOperationTrackerFactory.java | 4 +-
.../service/logging/LogManagerWithReplication.java | 19 +++-
asterixdb/pom.xml | 4 +-
.../appended-resources/supplemental-models.xml | 97 +++++++++----------
....com_netty_netty_netty-4.1.94.Final_NOTICE.txt} | 0
...ent.com_xerial_snappy-java_v1.1.10.1_NOTICE.txt | 16 ++++
.../apache/hyracks/api/comm/NetworkAddress.java | 7 +-
.../api/network/INetworkSecurityConfig.java | 9 ++
.../org/apache/hyracks/api/util/InvokeUtil.java | 40 +++++++-
.../hyracks/control/cc/work/CancelJobWork.java | 5 +
.../cc/work/JobletCleanupNotificationWork.java | 7 +-
.../hyracks/control/cc/work/RegisterNodeWork.java | 6 +-
.../control/cc/work/WaitForJobCompletionWork.java | 5 +
.../hyracks/control/nc/NodeControllerService.java | 2 +-
.../hyracks/control/nc/work/AbortTasksWork.java | 5 +
.../hyracks/control/nc/work/CleanupJobletWork.java | 7 +-
.../hyracks/control/nc/work/StartTasksWork.java | 5 +
.../hyracks/http/server/ChunkedResponse.java | 57 +++++++----
.../ipc/security/NetworkSecurityConfig.java | 7 ++
.../impls/AbstractAsynchronousScheduler.java | 52 +++++++++-
.../org/apache/hyracks/util/IOThrowingAction.java | 29 ++----
hyracks-fullstack/pom.xml | 25 ++++-
73 files changed, 709 insertions(+), 291 deletions(-)
diff --cc asterixdb/asterix-app/src/main/resources/cc-cloud-storage.conf
index dc6e5a2075,a61ac87783..be9bd544ea
--- a/asterixdb/asterix-app/src/main/resources/cc-cloud-storage.conf
+++ b/asterixdb/asterix-app/src/main/resources/cc-cloud-storage.conf
@@@ -65,3 -66,6 +64,5 @@@ compiler.sort.parallel=fals
compiler.internal.sanitycheck=true
messaging.frame.size=4096
messaging.frame.count=512
+ cloud.deployment=true
+ storage.buffercache.pagesize=32KB
->>>>>>> CHANGE (337517 [NO ISSUE][OTH] Move page size storage config to common conf)
diff --cc asterixdb/asterix-app/src/main/resources/cc_static_partitioning.conf
index dc6e5a2075,e3e779e17d..3923222c56
--- a/asterixdb/asterix-app/src/main/resources/cc_static_partitioning.conf
+++ b/asterixdb/asterix-app/src/main/resources/cc_static_partitioning.conf
@@@ -65,3 -66,6 +64,5 @@@ compiler.sort.parallel=fals
compiler.internal.sanitycheck=true
messaging.frame.size=4096
messaging.frame.count=512
+ storage.buffercache.pagesize=32KB
+ storage.partitioning=static
->>>>>>> CHANGE (337517 [NO ISSUE][OTH] Move page size storage config to common conf)
diff --cc asterixdb/asterix-app/src/test/resources/cc-analyze.conf
index 9532849649,d8b35122a1..16fa0aeea4
--- a/asterixdb/asterix-app/src/test/resources/cc-analyze.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-analyze.conf
@@@ -53,11 -54,6 +52,5 @@@ credential.file=src/test/resources/secu
[common]
log.dir = logs/
log.level = INFO
- compiler.framesize=32KB
- compiler.sortmemory=320KB
- compiler.groupmemory=160KB
- compiler.joinmemory=256KB
- compiler.textsearchmemory=160KB
- compiler.windowmemory=192KB
- messaging.frame.size=4096
- messaging.frame.count=512
+ compiler.groupmemory=64MB
+ storage.buffercache.pagesize=32KB
->>>>>>> CHANGE (337517 [NO ISSUE][OTH] Move page size storage config to common conf)
diff --cc asterixdb/asterix-app/src/test/resources/cc-batch-lookup.conf
index 6e10481dc4,18627b4e83..7d3aa5241b
--- a/asterixdb/asterix-app/src/test/resources/cc-batch-lookup.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-batch-lookup.conf
@@@ -59,7 -60,8 +58,7 @@@ compiler.groupmemory=160K
compiler.joinmemory=256KB
compiler.textsearchmemory=160KB
compiler.windowmemory=192KB
+compiler.batch.lookup=true
messaging.frame.size=4096
messaging.frame.count=512
-
-cloud.deployment=true
+ storage.buffercache.pagesize=32KB
->>>>>>> CHANGE (337517 [NO ISSUE][OTH] Move page size storage config to common conf)
diff --cc asterixdb/asterix-app/src/test/resources/cc-cloud-storage.conf
index 9532849649,18627b4e83..6373eaeec6
--- a/asterixdb/asterix-app/src/test/resources/cc-cloud-storage.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-cloud-storage.conf
@@@ -61,3 -62,6 +60,5 @@@ compiler.textsearchmemory=160K
compiler.windowmemory=192KB
messaging.frame.size=4096
messaging.frame.count=512
+ cloud.deployment=true
+ storage.buffercache.pagesize=32KB
->>>>>>> CHANGE (337517 [NO ISSUE][OTH] Move page size storage config to common conf)
diff --cc asterixdb/asterix-app/src/test/resources/cc-compression.conf
index 373d9bb770,373d9bb770..bc4ea4a728
--- a/asterixdb/asterix-app/src/test/resources/cc-compression.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-compression.conf
@@@ -35,7 -35,7 +35,6 @@@ address=127.0.0.
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
--storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
@@@ -58,3 -58,3 +57,4 @@@ compiler.windowmemory=192K
messaging.frame.size=4096
messaging.frame.count=512
storage.compression.block=snappy
++storage.buffercache.pagesize=32KB
diff --cc asterixdb/asterix-app/src/test/resources/cc-single-lookup.conf
index 6e10481dc4,1b006b3180..d59d4d58c5
--- a/asterixdb/asterix-app/src/test/resources/cc-single-lookup.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-single-lookup.conf
@@@ -59,7 -60,8 +58,7 @@@ compiler.groupmemory=160K
compiler.joinmemory=256KB
compiler.textsearchmemory=160KB
compiler.windowmemory=192KB
- compiler.batch.lookup=true
+ compiler.batch.lookup=false
messaging.frame.size=4096
messaging.frame.count=512
-
+ storage.buffercache.pagesize=32KB
->>>>>>> CHANGE (337517 [NO ISSUE][OTH] Move page size storage config to common conf)
[asterixdb] 16/20: Merge branch 'gerrit/cheshire-cat' into 'gerrit/neo'
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit c1a1742a1f546de015b1a65d78b3b559dc427876
Merge: 04b1dd4f41 1e5910e2df
Author: Michael Blow <mb...@apache.org>
AuthorDate: Tue Jul 25 11:03:07 2023 -0400
Merge branch 'gerrit/cheshire-cat' into 'gerrit/neo'
Change-Id: I7fcd4f954796bafa7f9452306c5b5e6eb3a540e6
.../appended-resources/supplemental-models.xml | 90 +++++++++++-----------
....com_netty_netty_netty-4.1.94.Final_NOTICE.txt} | 0
hyracks-fullstack/pom.xml | 17 +++-
3 files changed, 61 insertions(+), 46 deletions(-)
diff --cc hyracks-fullstack/pom.xml
index 843167a470,76f7b10441..7070ca605b
--- a/hyracks-fullstack/pom.xml
+++ b/hyracks-fullstack/pom.xml
@@@ -70,13 -70,13 +70,13 @@@
<test.includes>${global.test.includes}</test.includes>
<test.excludes>${global.test.excludes}</test.excludes>
<!-- Versions under dependencymanagement or used in many projects via properties -->
- <hadoop.version>2.8.5</hadoop.version>
+ <hadoop.version>3.3.1</hadoop.version>
<jacoco.version>0.7.6.201602180812</jacoco.version>
<log4j.version>2.19.0</log4j.version>
- <snappy.version>1.1.8.4</snappy.version>
- <jackson.version>2.14.1</jackson.version>
+ <snappy.version>1.1.10.1</snappy.version>
+ <jackson.version>2.14.3</jackson.version>
<jackson-databind.version>${jackson.version}</jackson-databind.version>
- <netty.version>4.1.87.Final</netty.version>
+ <netty.version>4.1.94.Final</netty.version>
<implementation.title>Apache Hyracks and Algebricks - ${project.name}</implementation.title>
<implementation.url>https://asterixdb.apache.org/</implementation.url>
@@@ -110,6 -110,6 +110,16 @@@
<artifactId>netty-codec</artifactId>
<version>${netty.version}</version>
</dependency>
++ <dependency>
++ <groupId>io.netty</groupId>
++ <artifactId>netty-codec-dns</artifactId>
++ <version>${netty.version}</version>
++ </dependency>
++ <dependency>
++ <groupId>io.netty</groupId>
++ <artifactId>netty-codec-socks</artifactId>
++ <version>${netty.version}</version>
++ </dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-handler</artifactId>
@@@ -125,6 -125,6 +135,11 @@@
<artifactId>netty-transport-classes-epoll</artifactId>
<version>${netty.version}</version>
</dependency>
++ <dependency>
++ <groupId>io.netty</groupId>
++ <artifactId>netty-resolver</artifactId>
++ <version>${netty.version}</version>
++ </dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-resolver-dns</artifactId>
[asterixdb] 07/20: [ASTERIXDB-3220][REPL] Allow waiting for IO on specific dataset partition
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit cc7d89ccbde59c209fdbe8a819c95fa567ea9224
Author: Murtadha Hubail <mh...@apache.org>
AuthorDate: Sun Jul 9 08:28:45 2023 -0700
[ASTERIXDB-3220][REPL] Allow waiting for IO on specific dataset partition
- user model changes: no
- storage format changes: no
- interface changes: yes
Details:
- Add API to wait for IO on a specific dataset partition.
- When waiting for a partition replica IO ops to finish, only wait
for the replica partition rather than all partitions.
Change-Id: I90f311f602b3c8526556f64d7b25672981fac320
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17633
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Ali Alsuliman <al...@gmail.com>
---
.../common/api/IDatasetLifecycleManager.java | 6 +++--
.../common/context/BaseOperationTracker.java | 8 +++---
.../apache/asterix/common/context/DatasetInfo.java | 31 ++++++++++++++++++++--
.../common/context/DatasetLifecycleManager.java | 4 +--
.../context/PrimaryIndexOperationTracker.java | 4 +--
.../ioopcallbacks/LSMIOOperationCallback.java | 6 +++--
.../ioopcallbacks/LSMIOOperationCallbackTest.java | 7 +++++
.../replication/sync/ReplicaSynchronizer.java | 2 +-
.../SecondaryIndexOperationTrackerFactory.java | 4 ++-
9 files changed, 56 insertions(+), 16 deletions(-)
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IDatasetLifecycleManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IDatasetLifecycleManager.java
index 1c2a047d86..c7eee2163e 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IDatasetLifecycleManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IDatasetLifecycleManager.java
@@ -161,12 +161,14 @@ public interface IDatasetLifecycleManager extends IResourceLifecycleManager<IInd
void flushDataset(IReplicationStrategy replicationStrategy, IntPredicate partitions) throws HyracksDataException;
/**
- * Waits for all ongoing IO operations on all open datasets that are matching {@code replicationStrategy}.
+ * Waits for all ongoing IO operations on all open datasets that are matching {@code replicationStrategy} and
+ * {@code partition}.
*
* @param replicationStrategy
+ * @param partition
* @throws HyracksDataException
*/
- void waitForIO(IReplicationStrategy replicationStrategy) throws HyracksDataException;
+ void waitForIO(IReplicationStrategy replicationStrategy, int partition) throws HyracksDataException;
/**
* @return the current datasets io stats
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/BaseOperationTracker.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/BaseOperationTracker.java
index 5964bb473c..7d3dba4ead 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/BaseOperationTracker.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/BaseOperationTracker.java
@@ -33,17 +33,19 @@ public class BaseOperationTracker implements ITransactionOperationTracker {
private static final Logger LOGGER = LogManager.getLogger();
protected final int datasetID;
protected final DatasetInfo dsInfo;
+ protected final int partition;
- public BaseOperationTracker(int datasetID, DatasetInfo dsInfo) {
+ public BaseOperationTracker(int datasetID, DatasetInfo dsInfo, int partition) {
this.datasetID = datasetID;
this.dsInfo = dsInfo;
+ this.partition = partition;
}
@Override
public void beforeOperation(ILSMIndex index, LSMOperationType opType, ISearchOperationCallback searchCallback,
IModificationOperationCallback modificationCallback) throws HyracksDataException {
if (opType == LSMOperationType.REPLICATE) {
- dsInfo.declareActiveIOOperation(REPLICATE);
+ dsInfo.declareActiveIOOperation(REPLICATE, partition);
}
}
@@ -59,7 +61,7 @@ public class BaseOperationTracker implements ITransactionOperationTracker {
public void completeOperation(ILSMIndex index, LSMOperationType opType, ISearchOperationCallback searchCallback,
IModificationOperationCallback modificationCallback) throws HyracksDataException {
if (opType == LSMOperationType.REPLICATE) {
- dsInfo.undeclareActiveIOOperation(REPLICATE);
+ dsInfo.undeclareActiveIOOperation(REPLICATE, partition);
}
}
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetInfo.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetInfo.java
index d15d9be927..87a3c2f02f 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetInfo.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetInfo.java
@@ -33,12 +33,16 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import it.unimi.dsi.fastutil.ints.Int2IntMap;
+import it.unimi.dsi.fastutil.ints.Int2IntOpenHashMap;
+
public class DatasetInfo extends Info implements Comparable<DatasetInfo> {
private static final Logger LOGGER = LogManager.getLogger();
// partition -> index
private final Map<Integer, Set<IndexInfo>> partitionIndexes;
// resourceID -> index
private final Map<Long, IndexInfo> indexes;
+ private final Int2IntMap partitionPendingIO;
private final int datasetID;
private final ILogManager logManager;
private final LogRecord waitLog = new LogRecord();
@@ -54,6 +58,7 @@ public class DatasetInfo extends Info implements Comparable<DatasetInfo> {
public DatasetInfo(int datasetID, ILogManager logManager) {
this.partitionIndexes = new HashMap<>();
this.indexes = new HashMap<>();
+ this.partitionPendingIO = new Int2IntOpenHashMap();
this.setLastAccess(-1);
this.datasetID = datasetID;
this.setRegistered(false);
@@ -74,7 +79,8 @@ public class DatasetInfo extends Info implements Comparable<DatasetInfo> {
setLastAccess(System.currentTimeMillis());
}
- public synchronized void declareActiveIOOperation(ILSMIOOperation.LSMIOOperationType opType) {
+ public synchronized void declareActiveIOOperation(ILSMIOOperation.LSMIOOperationType opType, int partition) {
+ partitionPendingIO.put(partition, partitionPendingIO.getOrDefault(partition, 0) + 1);
numActiveIOOps++;
switch (opType) {
case FLUSH:
@@ -91,7 +97,8 @@ public class DatasetInfo extends Info implements Comparable<DatasetInfo> {
}
}
- public synchronized void undeclareActiveIOOperation(ILSMIOOperation.LSMIOOperationType opType) {
+ public synchronized void undeclareActiveIOOperation(ILSMIOOperation.LSMIOOperationType opType, int partition) {
+ partitionPendingIO.put(partition, partitionPendingIO.getOrDefault(partition, 0) - 1);
numActiveIOOps--;
switch (opType) {
case FLUSH:
@@ -253,6 +260,26 @@ public class DatasetInfo extends Info implements Comparable<DatasetInfo> {
}
}
+ public void waitForIO(int partition) throws HyracksDataException {
+ logManager.log(waitLog);
+ synchronized (this) {
+ while (partitionPendingIO.getOrDefault(partition, 0) > 0) {
+ try {
+ wait();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw HyracksDataException.create(e);
+ }
+ }
+ if (partitionPendingIO.getOrDefault(partition, 0) < 0) {
+ LOGGER.error("number of IO operations cannot be negative for dataset {}, partition {}", this,
+ partition);
+ throw new IllegalStateException(
+ "Number of IO operations cannot be negative: " + this + ", partition " + partition);
+ }
+ }
+ }
+
public synchronized int getPendingFlushes() {
return pendingFlushes;
}
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
index 117b4fc4d0..4fc9dd6728 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
@@ -552,10 +552,10 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
}
@Override
- public void waitForIO(IReplicationStrategy replicationStrategy) throws HyracksDataException {
+ public void waitForIO(IReplicationStrategy replicationStrategy, int partition) throws HyracksDataException {
for (DatasetResource dsr : datasets.values()) {
if (dsr.isOpen() && replicationStrategy.isMatch(dsr.getDatasetID())) {
- dsr.getDatasetInfo().waitForIO();
+ dsr.getDatasetInfo().waitForIO(partition);
}
}
}
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
index b0d8e02af7..a4ad7cf232 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
@@ -59,7 +59,6 @@ import org.apache.logging.log4j.Logger;
@NotThreadSafe
public class PrimaryIndexOperationTracker extends BaseOperationTracker implements IoOperationCompleteListener {
private static final Logger LOGGER = LogManager.getLogger();
- private final int partition;
// Number of active operations on an ILSMIndex instance.
private final AtomicInteger numActiveOperations;
private final ILogManager logManager;
@@ -71,8 +70,7 @@ public class PrimaryIndexOperationTracker extends BaseOperationTracker implement
public PrimaryIndexOperationTracker(int datasetID, int partition, ILogManager logManager, DatasetInfo dsInfo,
ILSMComponentIdGenerator idGenerator) {
- super(datasetID, dsInfo);
- this.partition = partition;
+ super(datasetID, dsInfo, partition);
this.logManager = logManager;
this.numActiveOperations = new AtomicInteger();
this.idGenerator = idGenerator;
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java
index 1189b51960..f56e5c0a16 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMIOOperationCallback.java
@@ -74,6 +74,7 @@ public class LSMIOOperationCallback implements ILSMIOOperationCallback {
private final IIndexCheckpointManagerProvider indexCheckpointManagerProvider;
protected final DatasetInfo dsInfo;
protected final ILSMIndex lsmIndex;
+ private final int partition;
private long firstLsnForCurrentMemoryComponent = 0L;
private long persistenceLsn = 0L;
private int pendingFlushes = 0;
@@ -84,6 +85,7 @@ public class LSMIOOperationCallback implements ILSMIOOperationCallback {
this.dsInfo = dsInfo;
this.lsmIndex = lsmIndex;
this.indexCheckpointManagerProvider = indexCheckpointManagerProvider;
+ this.partition = ResourceReference.ofIndex(lsmIndex.getIndexIdentifier()).getPartitionNum();
componentIds.add(componentId);
}
@@ -259,7 +261,7 @@ public class LSMIOOperationCallback implements ILSMIOOperationCallback {
@Override
public synchronized void scheduled(ILSMIOOperation operation) throws HyracksDataException {
- dsInfo.declareActiveIOOperation(operation.getIOOpertionType());
+ dsInfo.declareActiveIOOperation(operation.getIOOpertionType(), partition);
if (operation.getIOOpertionType() == LSMIOOperationType.FLUSH) {
pendingFlushes++;
FlushOperation flush = (FlushOperation) operation;
@@ -282,7 +284,7 @@ public class LSMIOOperationCallback implements ILSMIOOperationCallback {
pendingFlushes == 0 ? firstLsnForCurrentMemoryComponent : (Long) map.get(KEY_FLUSH_LOG_LSN);
}
}
- dsInfo.undeclareActiveIOOperation(operation.getIOOpertionType());
+ dsInfo.undeclareActiveIOOperation(operation.getIOOpertionType(), partition);
}
public synchronized boolean hasPendingFlush() {
diff --git a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMIOOperationCallbackTest.java b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMIOOperationCallbackTest.java
index db0911bc17..33d513f7e4 100644
--- a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMIOOperationCallbackTest.java
+++ b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMIOOperationCallbackTest.java
@@ -77,6 +77,7 @@ public class LSMIOOperationCallbackTest extends TestCase {
String indexId = "mockIndexId";
Mockito.when(mockIndex.getNumberOfAllMemoryComponents()).thenReturn(numMemoryComponents);
Mockito.when(mockIndex.getCurrentMemoryComponent()).thenReturn(Mockito.mock(AbstractLSMMemoryComponent.class));
+ Mockito.when(mockIndex.getIndexIdentifier()).thenReturn(getIndexPath());
DatasetInfo dsInfo = new DatasetInfo(101, null);
LSMComponentIdGenerator idGenerator = new LSMComponentIdGenerator(numMemoryComponents, MIN_VALID_COMPONENT_ID);
LSMIOOperationCallback callback = new LSMIOOperationCallback(dsInfo, mockIndex, idGenerator.getId(),
@@ -140,6 +141,7 @@ public class LSMIOOperationCallbackTest extends TestCase {
ILSMComponentIdGenerator idGenerator = new LSMComponentIdGenerator(numMemoryComponents, MIN_VALID_COMPONENT_ID);
ILSMIndex mockIndex = Mockito.mock(ILSMIndex.class);
Mockito.when(mockIndex.getNumberOfAllMemoryComponents()).thenReturn(numMemoryComponents);
+ Mockito.when(mockIndex.getIndexIdentifier()).thenReturn(getIndexPath());
ILSMMemoryComponent mockComponent = Mockito.mock(AbstractLSMMemoryComponent.class);
Mockito.when(mockIndex.getCurrentMemoryComponent()).thenReturn(mockComponent);
LSMIOOperationCallback callback = new LSMIOOperationCallback(dsInfo, mockIndex, idGenerator.getId(),
@@ -161,6 +163,7 @@ public class LSMIOOperationCallbackTest extends TestCase {
ILSMComponentIdGenerator idGenerator = new LSMComponentIdGenerator(numMemoryComponents, MIN_VALID_COMPONENT_ID);
ILSMIndex mockIndex = Mockito.mock(ILSMIndex.class);
Mockito.when(mockIndex.getNumberOfAllMemoryComponents()).thenReturn(numMemoryComponents);
+ Mockito.when(mockIndex.getIndexIdentifier()).thenReturn(getIndexPath());
ILSMMemoryComponent mockComponent = Mockito.mock(AbstractLSMMemoryComponent.class);
Mockito.when(mockIndex.getCurrentMemoryComponent()).thenReturn(mockComponent);
LSMIOOperationCallback callback = new LSMIOOperationCallback(dsInfo, mockIndex, idGenerator.getId(),
@@ -221,4 +224,8 @@ public class LSMIOOperationCallbackTest extends TestCase {
Mockito.doReturn(indexCheckpointManager).when(indexCheckpointManagerProvider).get(Mockito.any());
return indexCheckpointManagerProvider;
}
+
+ private static String getIndexPath() {
+ return "storage/partition_0/dataverse/dataset/0/index";
+ }
}
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaSynchronizer.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaSynchronizer.java
index 459ff01245..68ccd54e50 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaSynchronizer.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaSynchronizer.java
@@ -103,6 +103,6 @@ public class ReplicaSynchronizer {
private void waitForReplicatedDatasetsIO() throws HyracksDataException {
// wait for IO operations to ensure replicated datasets files won't change during replica sync
final IReplicationStrategy replStrategy = appCtx.getReplicationManager().getReplicationStrategy();
- appCtx.getDatasetLifecycleManager().waitForIO(replStrategy);
+ appCtx.getDatasetLifecycleManager().waitForIO(replStrategy, replica.getIdentifier().getPartition());
}
}
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexOperationTrackerFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexOperationTrackerFactory.java
index a104ae39fc..827b7131fc 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexOperationTrackerFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexOperationTrackerFactory.java
@@ -21,6 +21,7 @@ package org.apache.asterix.transaction.management.opcallbacks;
import org.apache.asterix.common.api.IDatasetLifecycleManager;
import org.apache.asterix.common.api.INcApplicationContext;
import org.apache.asterix.common.context.BaseOperationTracker;
+import org.apache.asterix.common.utils.StoragePathUtil;
import org.apache.hyracks.api.application.INCServiceContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.IJsonSerializable;
@@ -46,7 +47,8 @@ public class SecondaryIndexOperationTrackerFactory implements ILSMOperationTrack
public ILSMOperationTracker getOperationTracker(INCServiceContext ctx, IResource resource) {
IDatasetLifecycleManager dslcManager =
((INcApplicationContext) ctx.getApplicationContext()).getDatasetLifecycleManager();
- return new BaseOperationTracker(datasetId, dslcManager.getDatasetInfo(datasetId));
+ int partition = StoragePathUtil.getPartitionNumFromRelativePath(resource.getPath());
+ return new BaseOperationTracker(datasetId, dslcManager.getDatasetInfo(datasetId), partition);
}
@Override
[asterixdb] 17/20: [NO ISSUE][*DB] Enable ability to configure RMI bind address
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 641462e6515cd0f0729f2a0a3bfe2986a5b5690e
Author: Michael Blow <mb...@apache.org>
AuthorDate: Thu Jul 27 17:00:36 2023 -0400
[NO ISSUE][*DB] Enable ability to configure RMI bind address
Change-Id: Ib0b759cbcbf6dc89e98ed378b3e44968356aaa90
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17659
Reviewed-by: Murtadha Al Hubail <mh...@apache.org>
Tested-by: Michael Blow <mb...@apache.org>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
---
.../apache/asterix/app/nc/NCAppRuntimeContext.java | 16 ++++----------
.../apache/asterix/metadata/RMIClientFactory.java | 13 ++++++++++-
.../apache/asterix/metadata/RMIServerFactory.java | 25 +++++++++++++++++++---
.../metadata/bootstrap/AsterixStateProxy.java | 15 +++----------
.../api/network/INetworkSecurityConfig.java | 9 ++++++++
.../ipc/security/NetworkSecurityConfig.java | 7 ++++++
6 files changed, 57 insertions(+), 28 deletions(-)
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
index 1a8916836d..d783f9950a 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
@@ -462,18 +462,10 @@ public class NCAppRuntimeContext implements INcApplicationContext {
final INetworkSecurityManager networkSecurityManager =
ncServiceContext.getControllerService().getNetworkSecurityManager();
- // clients need to have the client factory on their classpath- to enable older clients, only use
- // our client socket factory when SSL is enabled
- if (networkSecurityManager.getConfiguration().isSslEnabled()) {
- final RMIServerFactory serverSocketFactory = new RMIServerFactory(networkSecurityManager);
- final RMIClientFactory clientSocketFactory =
- new RMIClientFactory(networkSecurityManager.getConfiguration());
- metadataNodeStub = (IMetadataNode) UnicastRemoteObject.exportObject(MetadataNode.INSTANCE,
- getMetadataProperties().getMetadataPort(), clientSocketFactory, serverSocketFactory);
- } else {
- metadataNodeStub = (IMetadataNode) UnicastRemoteObject.exportObject(MetadataNode.INSTANCE,
- getMetadataProperties().getMetadataPort());
- }
+ metadataNodeStub = (IMetadataNode) UnicastRemoteObject.exportObject(MetadataNode.INSTANCE,
+ getMetadataProperties().getMetadataPort(),
+ RMIClientFactory.getSocketFactory(networkSecurityManager),
+ RMIServerFactory.getSocketFactory(networkSecurityManager));
}
}
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/RMIClientFactory.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/RMIClientFactory.java
index 515e763f58..ce459e27f7 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/RMIClientFactory.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/RMIClientFactory.java
@@ -29,6 +29,7 @@ import javax.net.SocketFactory;
import javax.net.ssl.SSLSocketFactory;
import org.apache.hyracks.api.network.INetworkSecurityConfig;
+import org.apache.hyracks.api.network.INetworkSecurityManager;
import org.apache.hyracks.ipc.security.NetworkSecurityManager;
public class RMIClientFactory implements RMIClientSocketFactory, Serializable {
@@ -37,11 +38,21 @@ public class RMIClientFactory implements RMIClientSocketFactory, Serializable {
private final INetworkSecurityConfig config;
private transient SocketFactory socketFactory;
- public RMIClientFactory(INetworkSecurityConfig config) {
+ private RMIClientFactory(INetworkSecurityConfig config) {
this.config = config;
}
+ public static RMIClientSocketFactory getSocketFactory(INetworkSecurityManager securityManager) {
+ // clients need to have the client factory on their classpath- to enable older clients, only use
+ // our client socket factory when SSL is enabled
+ INetworkSecurityConfig config = securityManager.getConfiguration();
+ if (config.isSslEnabled()) {
+ return new RMIClientFactory(config);
+ }
+ return null;
+ }
+
public Socket createSocket(String host, int port) throws IOException {
synchronized (this) {
if (socketFactory == null) {
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/RMIServerFactory.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/RMIServerFactory.java
index 9506c5a77e..0128a87d21 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/RMIServerFactory.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/RMIServerFactory.java
@@ -19,8 +19,10 @@
package org.apache.asterix.metadata;
import java.io.IOException;
+import java.net.InetAddress;
import java.net.ServerSocket;
import java.rmi.server.RMIServerSocketFactory;
+import java.util.Optional;
import javax.net.ServerSocketFactory;
@@ -28,17 +30,34 @@ import org.apache.hyracks.api.network.INetworkSecurityManager;
public class RMIServerFactory implements RMIServerSocketFactory {
+ // default backlog used by the JDK (e.g. sun.security.ssl.SSLServerSocketFactoryImpl)
+ private static final int DEFAULT_BACKLOG = 50;
private final INetworkSecurityManager securityManager;
- public RMIServerFactory(INetworkSecurityManager securityManager) {
+ private RMIServerFactory(INetworkSecurityManager securityManager) {
this.securityManager = securityManager;
}
+ public static RMIServerSocketFactory getSocketFactory(INetworkSecurityManager securityManager) {
+ if (securityManager.getConfiguration().isSslEnabled()) {
+ return new RMIServerFactory(securityManager);
+ }
+ return null;
+ }
+
@Override
public ServerSocket createServerSocket(int port) throws IOException {
+ ServerSocketFactory socketFactory;
if (securityManager.getConfiguration().isSslEnabled()) {
- return securityManager.newSSLContext().getServerSocketFactory().createServerSocket(port);
+ socketFactory = securityManager.newSSLContext().getServerSocketFactory();
+ } else {
+ socketFactory = ServerSocketFactory.getDefault();
+ }
+ Optional<InetAddress> rmiBindAddress = securityManager.getConfiguration().getRMIBindAddress();
+ if (rmiBindAddress.isPresent()) {
+ return socketFactory.createServerSocket(port, DEFAULT_BACKLOG, rmiBindAddress.get());
+ } else {
+ return socketFactory.createServerSocket(port);
}
- return ServerSocketFactory.getDefault().createServerSocket(port);
}
}
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/AsterixStateProxy.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/AsterixStateProxy.java
index 2104fdf1da..cedcccfca4 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/AsterixStateProxy.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/AsterixStateProxy.java
@@ -43,18 +43,9 @@ public class AsterixStateProxy implements IAsterixStateProxy {
public static IAsterixStateProxy registerRemoteObject(INetworkSecurityManager networkSecurityManager,
int metadataCallbackPort) throws RemoteException {
- IAsterixStateProxy stub;
- // clients need to have the client factory on their classpath- to enable older clients, only use
- // our client socket factory when SSL is enabled
- if (networkSecurityManager.getConfiguration().isSslEnabled()) {
- final RMIServerFactory serverSocketFactory = new RMIServerFactory(networkSecurityManager);
- final RMIClientFactory clientSocketFactory =
- new RMIClientFactory(networkSecurityManager.getConfiguration());
- stub = (IAsterixStateProxy) UnicastRemoteObject.exportObject(cc, metadataCallbackPort, clientSocketFactory,
- serverSocketFactory);
- } else {
- stub = (IAsterixStateProxy) UnicastRemoteObject.exportObject(cc, metadataCallbackPort);
- }
+ IAsterixStateProxy stub = (IAsterixStateProxy) UnicastRemoteObject.exportObject(cc, metadataCallbackPort,
+ RMIClientFactory.getSocketFactory(networkSecurityManager),
+ RMIServerFactory.getSocketFactory(networkSecurityManager));
LOGGER.info("Asterix Distributed State Proxy Bound");
return stub;
}
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/network/INetworkSecurityConfig.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/network/INetworkSecurityConfig.java
index b483158d12..7fc0335c5c 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/network/INetworkSecurityConfig.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/network/INetworkSecurityConfig.java
@@ -20,7 +20,9 @@ package org.apache.hyracks.api.network;
import java.io.File;
import java.io.Serializable;
+import java.net.InetAddress;
import java.security.KeyStore;
+import java.util.Optional;
public interface INetworkSecurityConfig extends Serializable {
@@ -65,4 +67,11 @@ public interface INetworkSecurityConfig extends Serializable {
* @return the trust store file
*/
File getTrustStoreFile();
+
+ /**
+ * The optional address to bind for RMI server sockets; or absent to bind to all addresses / interfaces.
+ *
+ * @return the optional bind address
+ */
+ Optional<InetAddress> getRMIBindAddress();
}
diff --git a/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/security/NetworkSecurityConfig.java b/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/security/NetworkSecurityConfig.java
index 2170c15365..bfcd623900 100644
--- a/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/security/NetworkSecurityConfig.java
+++ b/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/security/NetworkSecurityConfig.java
@@ -22,10 +22,12 @@ import java.io.File;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
+import java.net.InetAddress;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.cert.CertificateException;
+import java.util.Optional;
import org.apache.hyracks.api.network.INetworkSecurityConfig;
@@ -90,6 +92,11 @@ public class NetworkSecurityConfig implements INetworkSecurityConfig {
return trustStoreFile;
}
+ @Override
+ public Optional<InetAddress> getRMIBindAddress() {
+ return Optional.empty();
+ }
+
private void writeObject(ObjectOutputStream out) throws IOException {
out.defaultWriteObject();
writeStore(keyStore, out);
[asterixdb] 03/20: [NO ISSUE][*DB] Update dependencies
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 8b67de4caaf2455155c52896df4bb16f0591a73e
Author: Michael Blow <mb...@apache.org>
AuthorDate: Wed Jun 28 11:14:13 2023 -0400
[NO ISSUE][*DB] Update dependencies
- Jetty: 9.4.48.v20220622 -> 9.4.51.v20230217
- Jackson: 2.14.1 -> 2.14.3
- Snappy-Java: 1.1.8.4 -> 1.1.10.1
Change-Id: I4cd3cf27423c23cfbb0b5528026ebcf5840d7539
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17621
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Michael Blow <mb...@apache.org>
Tested-by: Michael Blow <mb...@apache.org>
---
.../org/apache/asterix/test/common/ResultExtractor.java | 4 ++--
asterixdb/pom.xml | 4 ++--
.../src/main/appended-resources/supplemental-models.xml | 7 ++++---
...ercontent.com_xerial_snappy-java_v1.1.10.1_NOTICE.txt | 16 ++++++++++++++++
hyracks-fullstack/pom.xml | 8 ++++----
5 files changed, 28 insertions(+), 11 deletions(-)
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ResultExtractor.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ResultExtractor.java
index 091fc20408..05b8bbfdea 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ResultExtractor.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ResultExtractor.java
@@ -214,8 +214,8 @@ public class ResultExtractor {
result = OBJECT_READER.readValue(resultStr);
} catch (Exception e) {
// whoops, not JSON (e.g. 404) - just include the body
- GlobalConfig.ASTERIX_LOGGER.log(Level.ERROR, resultStr);
- throw new Exception(resultStr);
+ GlobalConfig.ASTERIX_LOGGER.log(Level.ERROR, "result: {} json parse exception: {}", resultStr, e);
+ throw new Exception(resultStr, e);
}
final boolean isJsonFormat = isJsonFormat(fmt);
diff --git a/asterixdb/pom.xml b/asterixdb/pom.xml
index 33ca4a80d9..9181f142d5 100644
--- a/asterixdb/pom.xml
+++ b/asterixdb/pom.xml
@@ -1974,12 +1974,12 @@
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-util</artifactId>
- <version>9.4.48.v20220622</version>
+ <version>9.4.51.v20230217</version>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-util-ajax</artifactId>
- <version>9.4.48.v20220622</version>
+ <version>9.4.51.v20230217</version>
</dependency>
</dependencies>
</dependencyManagement>
diff --git a/asterixdb/src/main/appended-resources/supplemental-models.xml b/asterixdb/src/main/appended-resources/supplemental-models.xml
index d33451f77e..740cba9b34 100644
--- a/asterixdb/src/main/appended-resources/supplemental-models.xml
+++ b/asterixdb/src/main/appended-resources/supplemental-models.xml
@@ -463,9 +463,10 @@
<properties>
<!-- snappy-java is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
<!-- license override not needed, ALv2 is specified in its pom.xml -->
- <!-- see https://raw.githubusercontent.com/xerial/snappy-java/1.1.8.4/LICENSE -->
- <license.ignoreMissingEmbeddedLicense>1.1.8.4</license.ignoreMissingEmbeddedLicense>
- <license.ignoreMissingEmbeddedNotice>1.1.8.4</license.ignoreMissingEmbeddedNotice>
+ <!-- see https://raw.githubusercontent.com/xerial/snappy-java/v1.1.10.1/LICENSE -->
+ <license.ignoreMissingEmbeddedLicense>1.1.10.1</license.ignoreMissingEmbeddedLicense>
+ <license.ignoreMissingEmbeddedNotice>1.1.10.1</license.ignoreMissingEmbeddedNotice>
+ <license.ignoreNoticeOverride>1.1.10.1</license.ignoreNoticeOverride>
</properties>
</project>
</supplement>
diff --git a/asterixdb/src/main/licenses/content/raw.githubusercontent.com_xerial_snappy-java_v1.1.10.1_NOTICE.txt b/asterixdb/src/main/licenses/content/raw.githubusercontent.com_xerial_snappy-java_v1.1.10.1_NOTICE.txt
new file mode 100644
index 0000000000..19301705e1
--- /dev/null
+++ b/asterixdb/src/main/licenses/content/raw.githubusercontent.com_xerial_snappy-java_v1.1.10.1_NOTICE.txt
@@ -0,0 +1,16 @@
+This product includes software developed by Google
+ Snappy: http://code.google.com/p/snappy/ (New BSD License)
+
+This product includes software developed by Apache
+ PureJavaCrc32C from apache-hadoop-common http://hadoop.apache.org/
+ (Apache 2.0 license)
+
+This library containd statically linked libstdc++. This inclusion is allowed by
+"GCC RUntime Library Exception"
+http://gcc.gnu.org/onlinedocs/libstdc++/manual/license.html
+
+== Contributors ==
+ * Tatu Saloranta
+ * Providing benchmark suite
+ * Alec Wysoker
+ * Performance and memory usage improvement
diff --git a/hyracks-fullstack/pom.xml b/hyracks-fullstack/pom.xml
index 3d57c15c28..843167a470 100644
--- a/hyracks-fullstack/pom.xml
+++ b/hyracks-fullstack/pom.xml
@@ -73,8 +73,8 @@
<hadoop.version>3.3.1</hadoop.version>
<jacoco.version>0.7.6.201602180812</jacoco.version>
<log4j.version>2.19.0</log4j.version>
- <snappy.version>1.1.8.4</snappy.version>
- <jackson.version>2.14.1</jackson.version>
+ <snappy.version>1.1.10.1</snappy.version>
+ <jackson.version>2.14.3</jackson.version>
<jackson-databind.version>${jackson.version}</jackson-databind.version>
<netty.version>4.1.87.Final</netty.version>
@@ -482,12 +482,12 @@
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-util</artifactId>
- <version>9.4.48.v20220622</version>
+ <version>9.4.51.v20230217</version>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-util-ajax</artifactId>
- <version>9.4.48.v20220622</version>
+ <version>9.4.51.v20230217</version>
</dependency>
<!-- Manually included to avoid CVE-2023-1370 -->
<dependency>
[asterixdb] 20/20: Merge branch 'gerrit/trinity' into 'master'
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 89c10eb2b1a8393c438af494873304ae392e2e43
Merge: 1cb3eed1da 0f3fb3b0b6
Author: Michael Blow <mi...@couchbase.com>
AuthorDate: Mon Jul 31 22:24:45 2023 -0400
Merge branch 'gerrit/trinity' into 'master'
Change-Id: Ia4c6bdfa9dce9437049d59e137494b2372348754
.../http/server/NCQueryCancellationServlet.java | 3 +
.../api/http/server/NCQueryServiceServlet.java | 3 +-
.../app/active/ActiveEntityEventsListener.java | 10 +-
.../app/active/ActiveNotificationHandler.java | 52 +++++-----
.../asterix/app/message/CancelQueryRequest.java | 8 +-
.../message/ExecuteStatementRequestMessage.java | 4 +-
.../org/apache/asterix/app/nc/HaltCallback.java | 4 +-
.../apache/asterix/app/nc/NCAppRuntimeContext.java | 16 +---
.../asterix/app/translator/QueryTranslator.java | 4 +
.../rebalance/IDatasetRebalanceCallback.java | 8 +-
.../rebalance/NoOpDatasetRebalanceCallback.java | 3 +-
.../org/apache/asterix/utils/RebalanceUtil.java | 42 +++++---
...tic_partitioning.conf => cc-cloud-storage.conf} | 5 +-
.../src/main/resources/cc_static_partitioning.conf | 6 +-
.../asterix/test/common/ResultExtractor.java | 4 +-
.../asterix-app/src/test/resources/cc-analyze.conf | 2 +-
.../src/test/resources/cc-cloud-storage.conf | 7 --
.../common/api/IDatasetLifecycleManager.java | 6 +-
.../common/context/BaseOperationTracker.java | 8 +-
.../apache/asterix/common/context/DatasetInfo.java | 31 +++++-
.../common/context/DatasetLifecycleManager.java | 4 +-
.../context/PrimaryIndexOperationTracker.java | 11 +--
.../ioopcallbacks/LSMIOOperationCallback.java | 6 +-
.../ioopcallbacks/LSMIOOperationCallbackTest.java | 7 ++
.../apache/asterix/metadata/RMIClientFactory.java | 13 ++-
.../apache/asterix/metadata/RMIServerFactory.java | 25 ++++-
.../metadata/bootstrap/AsterixStateProxy.java | 15 +--
.../asterix/metadata/entities/NodeGroup.java | 3 +-
.../apache/asterix/metadata/utils/DatasetUtil.java | 2 +-
.../replication/api/ReplicationDestination.java | 32 +++++++
.../management/IndexReplicationManager.java | 99 +++++++++----------
.../management/ReplicationOperation.java | 106 +++++++++++++++++++++
.../replication/sync/ReplicaSynchronizer.java | 2 +-
.../SecondaryIndexOperationTrackerFactory.java | 4 +-
.../service/logging/LogManagerWithReplication.java | 19 +++-
asterixdb/pom.xml | 4 +-
.../appended-resources/supplemental-models.xml | 97 +++++++++----------
....com_netty_netty_netty-4.1.94.Final_NOTICE.txt} | 0
...ent.com_xerial_snappy-java_v1.1.10.1_NOTICE.txt | 16 ++++
.../apache/hyracks/api/comm/NetworkAddress.java | 7 +-
.../api/network/INetworkSecurityConfig.java | 9 ++
.../org/apache/hyracks/api/util/InvokeUtil.java | 40 +++++++-
.../hyracks/control/cc/work/CancelJobWork.java | 5 +
.../cc/work/JobletCleanupNotificationWork.java | 7 +-
.../hyracks/control/cc/work/RegisterNodeWork.java | 6 +-
.../control/cc/work/WaitForJobCompletionWork.java | 5 +
.../hyracks/control/nc/NodeControllerService.java | 2 +-
.../hyracks/control/nc/work/AbortTasksWork.java | 5 +
.../hyracks/control/nc/work/CleanupJobletWork.java | 7 +-
.../hyracks/control/nc/work/StartTasksWork.java | 5 +
.../hyracks/http/server/ChunkedResponse.java | 57 +++++++----
.../ipc/security/NetworkSecurityConfig.java | 7 ++
.../impls/AbstractAsynchronousScheduler.java | 52 +++++++++-
.../org/apache/hyracks/util/IOThrowingAction.java | 29 ++----
hyracks-fullstack/pom.xml | 25 ++++-
55 files changed, 690 insertions(+), 269 deletions(-)
diff --cc asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index 21ce833ac1,9f6a5cbf8e..d41b5499f3
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@@ -4796,23 -4860,11 +4796,27 @@@ public class QueryTranslator extends Ab
appCtx.getReceptionist().ensureSchedulable(schedulableRequest);
// ensure request not cancelled before running job
ensureNotCancelled(clientRequest);
- final JobId jobId = JobUtils.runJob(hcc, jobSpec, jobFlags, false);
+ if (atomicStatement != null) {
+ Dataset ds = metadataProvider.findDataset(((InsertStatement) atomicStatement).getDataverseName(),
+ ((InsertStatement) atomicStatement).getDatasetName());
+ atomic = ds.isAtomic();
+ if (atomic) {
+ int numParticipatingNodes = appCtx.getNodeJobTracker()
+ .getJobParticipatingNodes(jobSpec, LSMTreeIndexInsertUpdateDeleteOperatorDescriptor.class)
+ .size();
+ int numParticipatingPartitions = appCtx.getNodeJobTracker().getNumParticipatingPartitions(jobSpec,
+ LSMTreeIndexInsertUpdateDeleteOperatorDescriptor.class);
+ List<Integer> participatingDatasetIds = new ArrayList<>();
+ participatingDatasetIds.add(ds.getDatasetId());
+ jobSpec.setProperty(GlobalTxManager.GlOBAL_TX_PROPERTY_NAME, new GlobalTxInfo(
+ participatingDatasetIds, numParticipatingNodes, numParticipatingPartitions));
+ }
+ }
+ jobId = JobUtils.runJob(hcc, jobSpec, jobFlags, false);
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("createAndRunJob jobId:{}, uuid:{}", jobId,
+ requestParameters.getRequestReference().getUuid());
+ }
clientRequest.setJobId(jobId);
if (jId != null) {
jId.setValue(jobId);
diff --cc asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
index 1b0b4fdcd5,601cd0260e..837d2cb17a
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
@@@ -20,7 -20,8 +20,9 @@@ package org.apache.asterix.utils
import static org.apache.asterix.app.translator.QueryTranslator.abort;
import static org.apache.asterix.common.config.DatasetConfig.DatasetType;
+ import static org.apache.asterix.common.utils.IdentifierUtil.dataset;
+ import static org.apache.asterix.metadata.utils.DatasetUtil.getFullyQualifiedDisplayName;
+import static org.apache.asterix.om.utils.ProjectionFiltrationTypeUtil.ALL_FIELDS_TYPE;
import static org.apache.hyracks.storage.am.common.dataflow.IndexDropOperatorDescriptor.DropOption;
import java.rmi.RemoteException;
@@@ -83,14 -82,20 +85,15 @@@ public class RebalanceUtil
/**
* Rebalances an existing dataset to a list of target nodes.
*
- * @param dataverseName,
- * the dataverse name.
- * @param datasetName,
- * the dataset name.
- * @param targetNcNames,
- * the list of target nodes.
- * @param metadataProvider,
- * the metadata provider.
- * @param hcc,
- * the reusable hyracks connection.
+ * @param dataverseName, the dataverse name.
+ * @param datasetName, the dataset name.
+ * @param targetNcNames, the list of target nodes.
+ * @param metadataProvider, the metadata provider.
+ * @param hcc, the reusable hyracks connection.
+ * @return <code>false</code> if the rebalance was safely skipped
* @throws Exception
*/
- public static void rebalance(DataverseName dataverseName, String datasetName, Set<String> targetNcNames,
+ public static boolean rebalance(DataverseName dataverseName, String datasetName, Set<String> targetNcNames,
MetadataProvider metadataProvider, IHyracksClientConnection hcc,
IDatasetRebalanceCallback datasetRebalanceCallback, boolean forceRebalance) throws Exception {
Dataset sourceDataset;
diff --cc asterixdb/asterix-app/src/main/resources/cc_static_partitioning.conf
index 08d9dfd177,3923222c56..9884116a32
--- a/asterixdb/asterix-app/src/main/resources/cc_static_partitioning.conf
+++ b/asterixdb/asterix-app/src/main/resources/cc_static_partitioning.conf
@@@ -62,4 -65,4 +65,5 @@@ compiler.internal.sanitycheck=tru
messaging.frame.size=4096
messaging.frame.count=512
storage.buffercache.pagesize=32KB
- storage.partitioning=static
+ storage.partitioning=static
++storage.compression.block=snappy
diff --cc asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
index 3ec2d22869,9c2b6564ab..2e32fb32a9
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
@@@ -75,13 -66,10 +73,12 @@@ public class PrimaryIndexOperationTrack
private boolean flushLogCreated = false;
private final Map<String, FlushOperation> scheduledFlushes = new HashMap<>();
private long lastFlushTime = System.nanoTime();
+ private final Map<String, FlushOperation> lastFlushOperation = new HashMap<>();
+ private final IIndexCheckpointManagerProvider indexCheckpointManagerProvider;
public PrimaryIndexOperationTracker(int datasetID, int partition, ILogManager logManager, DatasetInfo dsInfo,
- ILSMComponentIdGenerator idGenerator) {
+ ILSMComponentIdGenerator idGenerator, IIndexCheckpointManagerProvider indexCheckpointManagerProvider) {
- super(datasetID, dsInfo);
- this.partition = partition;
+ super(datasetID, dsInfo, partition);
this.logManager = logManager;
this.numActiveOperations = new AtomicInteger();
this.idGenerator = idGenerator;
diff --cc asterixdb/pom.xml
index 681e9a8236,c1ee08facf..30a987e7a7
--- a/asterixdb/pom.xml
+++ b/asterixdb/pom.xml
@@@ -2026,13 -1993,8 +2026,13 @@@
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-util-ajax</artifactId>
- <version>9.4.48.v20220622</version>
+ <version>9.4.51.v20230217</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ <version>1.11.1</version>
+ </dependency>
</dependencies>
</dependencyManagement>
[asterixdb] 02/20: [NO ISSUE][*DB] Provide ability for IDatasetRebalanceCallback to decline a rebalance
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 767596bb2676377e5d8e10b4409d390eae46deb0
Author: Michael Blow <mi...@couchbase.com>
AuthorDate: Wed Jun 21 13:24:53 2023 -0400
[NO ISSUE][*DB] Provide ability for IDatasetRebalanceCallback to decline a rebalance
Change-Id: I1b2cdb56d70ce47c7b15061cc5ecc4af36ed8f61
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17608
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Michael Blow <mb...@apache.org>
Reviewed-by: Murtadha Al Hubail <mh...@apache.org>
---
.../rebalance/IDatasetRebalanceCallback.java | 8 +++--
.../rebalance/NoOpDatasetRebalanceCallback.java | 3 +-
.../org/apache/asterix/utils/RebalanceUtil.java | 42 +++++++++++++++-------
.../asterix/metadata/entities/NodeGroup.java | 3 +-
.../apache/asterix/metadata/utils/DatasetUtil.java | 5 ++-
5 files changed, 42 insertions(+), 19 deletions(-)
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/rebalance/IDatasetRebalanceCallback.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/rebalance/IDatasetRebalanceCallback.java
index e8683c91c5..c366977ea0 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/rebalance/IDatasetRebalanceCallback.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/rebalance/IDatasetRebalanceCallback.java
@@ -30,7 +30,7 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
public interface IDatasetRebalanceCallback {
/**
- * The action to perform before the target dataset is populated.
+ * The check to perform before the target dataset is populated.
*
* @param metadataProvider,
* the metadata provider.
@@ -40,9 +40,13 @@ public interface IDatasetRebalanceCallback {
* the target dataset.
* @param hcc,
* the hyracks client connection.
+ *
+ * @return <code>true</code> if the rebalance of the dataset should proceed, otherwise <code>false</code> to skip.
+ * If the dataset is skipped, the active metadata transaction context, if any, can be expected to be aborted.
+ *
* @throws HyracksDataException
*/
- void beforeRebalance(MetadataProvider metadataProvider, Dataset source, Dataset target,
+ boolean canRebalance(MetadataProvider metadataProvider, Dataset source, Dataset target,
IHyracksClientConnection hcc) throws HyracksDataException;
/**
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/rebalance/NoOpDatasetRebalanceCallback.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/rebalance/NoOpDatasetRebalanceCallback.java
index 680adbfce4..7085567c50 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/rebalance/NoOpDatasetRebalanceCallback.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/rebalance/NoOpDatasetRebalanceCallback.java
@@ -33,9 +33,10 @@ public class NoOpDatasetRebalanceCallback implements IDatasetRebalanceCallback {
}
@Override
- public void beforeRebalance(MetadataProvider metadataProvider, Dataset source, Dataset target,
+ public boolean canRebalance(MetadataProvider metadataProvider, Dataset source, Dataset target,
IHyracksClientConnection hcc) {
// Does nothing.
+ return true;
}
@Override
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
index 61562d831f..601cd0260e 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
@@ -20,6 +20,8 @@ package org.apache.asterix.utils;
import static org.apache.asterix.app.translator.QueryTranslator.abort;
import static org.apache.asterix.common.config.DatasetConfig.DatasetType;
+import static org.apache.asterix.common.utils.IdentifierUtil.dataset;
+import static org.apache.asterix.metadata.utils.DatasetUtil.getFullyQualifiedDisplayName;
import static org.apache.hyracks.storage.am.common.dataflow.IndexDropOperatorDescriptor.DropOption;
import java.rmi.RemoteException;
@@ -90,13 +92,15 @@ public class RebalanceUtil {
* the metadata provider.
* @param hcc,
* the reusable hyracks connection.
+ * @return <code>false</code> if the rebalance was safely skipped
* @throws Exception
*/
- public static void rebalance(DataverseName dataverseName, String datasetName, Set<String> targetNcNames,
+ public static boolean rebalance(DataverseName dataverseName, String datasetName, Set<String> targetNcNames,
MetadataProvider metadataProvider, IHyracksClientConnection hcc,
IDatasetRebalanceCallback datasetRebalanceCallback, boolean forceRebalance) throws Exception {
Dataset sourceDataset;
Dataset targetDataset;
+ boolean success = true;
// Executes the first Metadata transaction.
// Generates the rebalance target files. While doing that, hold read locks on the dataset so
// that no one can drop the rebalance source dataset.
@@ -108,13 +112,13 @@ public class RebalanceUtil {
// If the source dataset doesn't exist, then it's a no-op.
if (sourceDataset == null) {
- return;
+ return true;
}
Set<String> sourceNodes = new HashSet<>(metadataProvider.findNodes(sourceDataset.getNodeGroupName()));
if (!forceRebalance && sourceNodes.equals(targetNcNames)) {
- return;
+ return true;
}
if (!targetNcNames.isEmpty()) {
@@ -125,20 +129,25 @@ public class RebalanceUtil {
// The target dataset for rebalance.
targetDataset = sourceDataset.getTargetDatasetForRebalance(nodeGroupName);
- LOGGER.info("Rebalancing dataset {} from node group {} with nodes {} to node group {} with nodes {}",
- sourceDataset.getDatasetName(), sourceDataset.getNodeGroupName(), sourceNodes,
- targetDataset.getNodeGroupName(), targetNcNames);
+ LOGGER.info("Rebalancing {} {} from node group {} with nodes {} to node group {} with nodes {}",
+ dataset(), getFullyQualifiedDisplayName(sourceDataset), sourceDataset.getNodeGroupName(),
+ sourceNodes, targetDataset.getNodeGroupName(), targetNcNames);
// Rebalances the source dataset into the target dataset.
if (sourceDataset.getDatasetType() != DatasetType.EXTERNAL) {
- rebalance(sourceDataset, targetDataset, metadataProvider, hcc, datasetRebalanceCallback);
+ success = rebalance(sourceDataset, targetDataset, metadataProvider, hcc, datasetRebalanceCallback);
}
} else {
targetDataset = null;
// if this the last NC in the cluster, just drop the dataset
purgeDataset(sourceDataset, metadataProvider, hcc);
}
- // Complete the metadata transaction.
- MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
+ if (success) {
+ // Complete the metadata transaction.
+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
+ } else {
+ // Abort the metadata transaction, since we failed to rebalance the dataset
+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
+ }
} catch (Exception e) {
abort(e, e, mdTxnCtx);
throw e;
@@ -146,7 +155,10 @@ public class RebalanceUtil {
if (targetNcNames.isEmpty()) {
// Nothing else to do since the dataset was dropped.
- return;
+ return true;
+ } else if (!success) {
+ LOGGER.info("Dataset {} rebalance was skipped, see above log for reason", datasetName);
+ return false;
}
// Up to this point, since the bulk part of a rebalance operation is done,
// the following two operations will retry after interrupt and finally rethrow InterruptedException,
@@ -165,6 +177,7 @@ public class RebalanceUtil {
runMetadataTransaction(metadataProvider, () -> dropSourceDataset(sourceDataset, metadataProvider, hcc));
});
LOGGER.info("Dataset {} rebalance completed successfully", datasetName);
+ return true;
}
@FunctionalInterface
@@ -214,13 +227,16 @@ public class RebalanceUtil {
}
// Rebalances from the source to the target.
- private static void rebalance(Dataset source, Dataset target, MetadataProvider metadataProvider,
+ private static boolean rebalance(Dataset source, Dataset target, MetadataProvider metadataProvider,
IHyracksClientConnection hcc, IDatasetRebalanceCallback datasetRebalanceCallback) throws Exception {
// Drops the target dataset files (if any) to make rebalance idempotent.
dropDatasetFiles(target, metadataProvider, hcc);
// Performs the specified operation before the target dataset is populated.
- datasetRebalanceCallback.beforeRebalance(metadataProvider, source, target, hcc);
+ if (!datasetRebalanceCallback.canRebalance(metadataProvider, source, target, hcc)) {
+ // the callback indicates that this rebalance should be skipped; short circuit the remaining steps
+ return false;
+ }
// Creates the rebalance target.
createRebalanceTarget(target, metadataProvider, hcc);
@@ -233,6 +249,8 @@ public class RebalanceUtil {
// Performs the specified operation after the target dataset is populated.
datasetRebalanceCallback.afterRebalance(metadataProvider, source, target, hcc);
+
+ return true;
}
// Switches the metadata entity from the source dataset to the target dataset.
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/entities/NodeGroup.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/entities/NodeGroup.java
index a8d027f90c..a50fd4a5ac 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/entities/NodeGroup.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/entities/NodeGroup.java
@@ -20,6 +20,7 @@
package org.apache.asterix.metadata.entities;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.List;
@@ -42,7 +43,7 @@ public class NodeGroup implements IMetadataEntity<NodeGroup> {
this.nodeNames = nodeNames;
}
- public static NodeGroup createOrdered(String groupName, List<String> nodeNames) {
+ public static NodeGroup createOrdered(String groupName, Collection<String> nodeNames) {
List<String> sortedNodeNames = new ArrayList<>(nodeNames);
Collections.sort(sortedNodeNames);
return new NodeGroup(groupName, sortedNodeNames);
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/DatasetUtil.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/DatasetUtil.java
index 6529794e96..b2e68173cc 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/DatasetUtil.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/DatasetUtil.java
@@ -21,7 +21,6 @@ package org.apache.asterix.metadata.utils;
import static org.apache.asterix.common.utils.IdentifierUtil.dataset;
import java.io.DataOutput;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -613,10 +612,10 @@ public class DatasetUtil {
appCtx.getMetadataLockManager().acquireNodeGroupWriteLock(metadataProvider.getLocks(), nodeGroup);
NodeGroup ng = MetadataManager.INSTANCE.getNodegroup(mdTxnCtx, nodeGroup);
if (ng != null) {
- nodeGroup = nodeGroup + "_" + UUID.randomUUID().toString();
+ nodeGroup = nodeGroup + "_" + UUID.randomUUID();
appCtx.getMetadataLockManager().acquireNodeGroupWriteLock(metadataProvider.getLocks(), nodeGroup);
}
- MetadataManager.INSTANCE.addNodegroup(mdTxnCtx, NodeGroup.createOrdered(nodeGroup, new ArrayList<>(ncNames)));
+ MetadataManager.INSTANCE.addNodegroup(mdTxnCtx, NodeGroup.createOrdered(nodeGroup, ncNames));
return nodeGroup;
}
[asterixdb] 13/20: [ASTERIXDB-3226][STO] Warn on missing primary index during flush
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit e64d04628930c3aac2d4b3d10a098f35e9ca0464
Author: Murtadha Hubail <mh...@apache.org>
AuthorDate: Tue Jul 18 17:22:24 2023 +0300
[ASTERIXDB-3226][STO] Warn on missing primary index during flush
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
- To account for leaked secondary files, warn rather than halting
on missing primary index during flush to allow recovery to complete.
Eventually, the leaked files will be deleted during global recovery.
- Fix log message.
Change-Id: I7858a67cfc97fb1fde01c463a256083952d51543
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17647
Reviewed-by: Michael Blow <mb...@apache.org>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
---
.../apache/asterix/app/message/ExecuteStatementRequestMessage.java | 2 +-
.../asterix/common/context/PrimaryIndexOperationTracker.java | 7 +++----
2 files changed, 4 insertions(+), 5 deletions(-)
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
index 9b722df12c..05bc87b2f5 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
@@ -244,6 +244,6 @@ public class ExecuteStatementRequestMessage implements ICcAddressedMessage {
@Override
public String toString() {
return String.format("%s(id=%s, from=%s, uuid=%s): %s", getClass().getSimpleName(), requestMessageId,
- requestNodeId, LogRedactionUtil.statement(statementsText), requestReference.getUuid());
+ requestNodeId, requestReference.getUuid(), LogRedactionUtil.statement(statementsText));
}
}
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
index b0d8e02af7..8491f6ccb9 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
@@ -51,7 +51,6 @@ import org.apache.hyracks.storage.am.lsm.common.impls.FlushOperation;
import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentId;
import org.apache.hyracks.storage.common.IModificationOperationCallback;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
-import org.apache.hyracks.util.ExitUtil;
import org.apache.hyracks.util.annotations.NotThreadSafe;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -151,10 +150,10 @@ public class PrimaryIndexOperationTracker extends BaseOperationTracker implement
}
}
if (primaryLsmIndex == null) {
- LOGGER.fatal(
- "Primary index not found in dataset {} and partition {} open indexes {}; halting to clear memory state",
+ LOGGER.warn(
+ "Primary index not found in dataset {} and partition {} open indexes {}; possible secondary index leaked files",
dsInfo.getDatasetID(), partition, indexes);
- ExitUtil.halt(ExitUtil.EC_INCONSISTENT_STORAGE_REFERENCES);
+ return;
}
for (ILSMIndex lsmIndex : indexes) {
ILSMOperationTracker opTracker = lsmIndex.getOperationTracker();
[asterixdb] 08/20: [ASTERIXDB-3221][REPL] Use IO scheduler for replication ops
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 5f076a8347df9a45634914c76b4117b38b5731a6
Author: Murtadha Hubail <mh...@apache.org>
AuthorDate: Mon Jul 10 16:38:47 2023 -0700
[ASTERIXDB-3221][REPL] Use IO scheduler for replication ops
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
- To allow concurrent replication operations, use the IO scheduler
to schedule the operations using the exact same scheduling logic
as the flush operations.
- Maximum concurrent replication operations = maximum concurrent flush
operations.
- Allow connections to replicas to be recycled and reused by the
replication operations and close them when no more pending operations.
- Do not halt on replicate operations failures since failures are
expected and replicas can be re-synced.
Change-Id: I82aeb60381f90a254ca99274f7e9a38f64bc7a46
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17635
Reviewed-by: Ali Alsuliman <al...@gmail.com>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
---
.../org/apache/asterix/app/nc/HaltCallback.java | 4 +-
.../replication/api/ReplicationDestination.java | 32 +++++++
.../management/IndexReplicationManager.java | 99 +++++++++----------
.../management/ReplicationOperation.java | 106 +++++++++++++++++++++
.../impls/AbstractAsynchronousScheduler.java | 52 +++++++++-
5 files changed, 241 insertions(+), 52 deletions(-)
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/HaltCallback.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/HaltCallback.java
index 98020015dc..bb689ac5d0 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/HaltCallback.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/HaltCallback.java
@@ -41,6 +41,8 @@ public class HaltCallback implements IIoOperationFailedCallback {
@Override
public void operationFailed(ILSMIOOperation operation, Throwable t) {
LOGGER.error("Operation {} has failed", operation, t);
- ExitUtil.halt(ExitUtil.EC_IO_OPERATION_FAILED);
+ if (operation.getIOOpertionType() != ILSMIOOperation.LSMIOOperationType.REPLICATE) {
+ ExitUtil.halt(ExitUtil.EC_IO_OPERATION_FAILED);
+ }
}
}
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/api/ReplicationDestination.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/api/ReplicationDestination.java
index d8037569e8..f7a739bb38 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/api/ReplicationDestination.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/api/ReplicationDestination.java
@@ -20,7 +20,10 @@ package org.apache.asterix.replication.api;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.util.ArrayDeque;
+import java.util.HashMap;
import java.util.HashSet;
+import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
@@ -29,6 +32,7 @@ import org.apache.asterix.common.api.INcApplicationContext;
import org.apache.asterix.common.exceptions.ReplicationException;
import org.apache.asterix.common.replication.IPartitionReplica;
import org.apache.asterix.common.replication.IReplicationDestination;
+import org.apache.asterix.common.storage.ReplicaIdentifier;
import org.apache.asterix.replication.management.NetworkingUtil;
import org.apache.asterix.replication.messaging.ReplicationProtocol;
import org.apache.hyracks.api.network.ISocketChannel;
@@ -41,6 +45,7 @@ public class ReplicationDestination implements IReplicationDestination {
private static final Logger LOGGER = LogManager.getLogger();
private final Set<IPartitionReplica> replicas = new HashSet<>();
private final InetSocketAddress inputLocation;
+ private final Map<ReplicaIdentifier, ArrayDeque<PartitionReplica>> replicasConnPool = new HashMap<>();
private InetSocketAddress resolvedLocation;
private ISocketChannel logRepChannel;
@@ -64,6 +69,11 @@ public class ReplicationDestination implements IReplicationDestination {
@Override
public synchronized void remove(IPartitionReplica replica) {
replicas.remove(replica);
+ ArrayDeque<PartitionReplica> partitionConnections = replicasConnPool.remove(replica.getIdentifier());
+ if (partitionConnections != null) {
+ partitionConnections.forEach(PartitionReplica::close);
+ partitionConnections.clear();
+ }
}
@Override
@@ -138,4 +148,26 @@ public class ReplicationDestination implements IReplicationDestination {
public int hashCode() {
return Objects.hash(inputLocation);
}
+
+ public synchronized PartitionReplica getPartitionReplicaConnection(ReplicaIdentifier identifier,
+ INcApplicationContext appCtx) {
+ ArrayDeque<PartitionReplica> partitionReplicas =
+ replicasConnPool.computeIfAbsent(identifier, k -> new ArrayDeque<>());
+ if (!partitionReplicas.isEmpty()) {
+ return partitionReplicas.remove();
+ }
+ return new PartitionReplica(identifier, appCtx);
+ }
+
+ public synchronized void recycleConnection(PartitionReplica partitionReplica) {
+ ArrayDeque<PartitionReplica> partitionReplicas = replicasConnPool.get(partitionReplica.getIdentifier());
+ if (partitionReplicas != null) {
+ partitionReplicas.add(partitionReplica);
+ }
+ }
+
+ public synchronized void closeConnections() {
+ replicasConnPool
+ .forEach(((identifier, partitionReplicas) -> partitionReplicas.forEach(PartitionReplica::close)));
+ }
}
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/IndexReplicationManager.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/IndexReplicationManager.java
index 063709aa10..8c514bd694 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/IndexReplicationManager.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/IndexReplicationManager.java
@@ -21,22 +21,21 @@ package org.apache.asterix.replication.management;
import java.util.HashSet;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.asterix.common.api.INcApplicationContext;
import org.apache.asterix.common.exceptions.ReplicationException;
-import org.apache.asterix.common.replication.IPartitionReplica;
import org.apache.asterix.common.replication.IReplicationDestination;
import org.apache.asterix.common.replication.IReplicationManager;
import org.apache.asterix.common.replication.IReplicationStrategy;
import org.apache.asterix.common.storage.DatasetResourceReference;
-import org.apache.asterix.common.storage.ResourceReference;
-import org.apache.asterix.replication.api.PartitionReplica;
import org.apache.asterix.replication.api.ReplicationDestination;
-import org.apache.asterix.replication.sync.IndexSynchronizer;
import org.apache.asterix.transaction.management.resource.PersistentLocalResourceRepository;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.replication.IReplicationJob;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationScheduler;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexReplicationJob;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -45,13 +44,15 @@ public class IndexReplicationManager {
private static final Logger LOGGER = LogManager.getLogger();
private final IReplicationManager replicationManager;
- private final Set<ReplicationDestination> destinations = new HashSet<>();
+ private final Set<ReplicationDestination> destinations = ConcurrentHashMap.newKeySet();
private final LinkedBlockingQueue<IReplicationJob> replicationJobsQ = new LinkedBlockingQueue<>();
private final IReplicationStrategy replicationStrategy;
private final PersistentLocalResourceRepository resourceRepository;
private final INcApplicationContext appCtx;
+ private final ILSMIOOperationScheduler ioScheduler;
private final Object transferLock = new Object();
private final Set<ReplicationDestination> failedDest = new HashSet<>();
+ private final AtomicInteger pendingRepOpsCount = new AtomicInteger();
public IndexReplicationManager(INcApplicationContext appCtx, IReplicationManager replicationManager) {
this.appCtx = appCtx;
@@ -59,6 +60,8 @@ public class IndexReplicationManager {
this.resourceRepository = (PersistentLocalResourceRepository) appCtx.getLocalResourceRepository();
replicationStrategy = replicationManager.getReplicationStrategy();
appCtx.getThreadExecutor().execute(new ReplicationJobsProcessor());
+ ioScheduler = appCtx.getStorageComponentProvider().getIoOperationSchedulerProvider()
+ .getIoScheduler(appCtx.getServiceContext());
}
public void register(ReplicationDestination dest) {
@@ -72,12 +75,18 @@ public class IndexReplicationManager {
public void unregister(IReplicationDestination dest) {
synchronized (transferLock) {
LOGGER.info(() -> "unregister " + dest);
+ for (ReplicationDestination existingDest : destinations) {
+ if (existingDest.equals(dest)) {
+ existingDest.closeConnections();
+ break;
+ }
+ }
destinations.remove(dest);
failedDest.remove(dest);
}
}
- private void handleFailure(ReplicationDestination dest, Exception e) {
+ public void handleFailure(ReplicationDestination dest, Exception e) {
synchronized (transferLock) {
if (failedDest.contains(dest)) {
return;
@@ -87,6 +96,7 @@ public class IndexReplicationManager {
LOGGER.error("replica at {} failed", dest);
failedDest.add(dest);
}
+ dest.closeConnections();
replicationManager.notifyFailure(dest, e);
}
}
@@ -99,71 +109,62 @@ public class IndexReplicationManager {
process(job);
}
+ public Set<ReplicationDestination> getDestinations() {
+ synchronized (transferLock) {
+ return destinations;
+ }
+ }
+
private void process(IReplicationJob job) {
- try {
- if (skip(job)) {
- return;
- }
- synchronized (transferLock) {
- if (destinations.isEmpty()) {
- return;
- }
- final IndexSynchronizer synchronizer = new IndexSynchronizer(job, appCtx);
- final int indexPartition = getJobPartition(job);
- for (ReplicationDestination dest : destinations) {
- try {
- Optional<IPartitionReplica> partitionReplica = dest.getPartitionReplica(indexPartition);
- if (!partitionReplica.isPresent()) {
- continue;
- }
- PartitionReplica replica = (PartitionReplica) partitionReplica.get();
- synchronizer.sync(replica);
- } catch (Exception e) {
- handleFailure(dest, e);
- }
- }
- closeChannels();
- }
- } finally {
+ pendingRepOpsCount.incrementAndGet();
+ Optional<DatasetResourceReference> jobIndexRefOpt = getJobIndexRef(job);
+ if (jobIndexRefOpt.isEmpty()) {
+ LOGGER.warn("skipping replication of {} due to missing dataset resource reference", job.getAnyFile());
afterReplication(job);
+ return;
+ }
+ ReplicationOperation rp = new ReplicationOperation(appCtx, jobIndexRefOpt.get(), job, this);
+ if (job.getExecutionType() == IReplicationJob.ReplicationExecutionType.SYNC) {
+ rp.call();
+ } else {
+ try {
+ ioScheduler.scheduleOperation(rp);
+ } catch (HyracksDataException e) {
+ throw new ReplicationException(e);
+ }
}
}
- private boolean skip(IReplicationJob job) {
+ public boolean skip(DatasetResourceReference indexRef) {
+ return !replicationStrategy.isMatch(indexRef.getDatasetId());
+ }
+
+ public Optional<DatasetResourceReference> getJobIndexRef(IReplicationJob job) {
+ final String fileToReplicate = job.getAnyFile();
try {
- final String fileToReplicate = job.getAnyFile();
- final Optional<DatasetResourceReference> indexFileRefOpt =
- resourceRepository.getLocalResourceReference(fileToReplicate);
- if (!indexFileRefOpt.isPresent()) {
- LOGGER.warn("skipping replication of {} due to missing dataset resource reference", fileToReplicate);
- return true;
- }
- return !replicationStrategy.isMatch(indexFileRefOpt.get().getDatasetId());
+ return resourceRepository.getLocalResourceReference(fileToReplicate);
} catch (HyracksDataException e) {
throw new IllegalStateException("Couldn't find resource for " + job.getAnyFile(), e);
}
}
- private int getJobPartition(IReplicationJob job) {
- return ResourceReference.of(job.getAnyFile()).getPartitionNum();
- }
-
private void closeChannels() {
- if (!replicationJobsQ.isEmpty()) {
- return;
- }
LOGGER.trace("no pending replication jobs; closing connections to replicas");
for (ReplicationDestination dest : destinations) {
- dest.getReplicas().stream().map(PartitionReplica.class::cast).forEach(PartitionReplica::close);
+ dest.closeConnections();
}
}
- private static void afterReplication(IReplicationJob job) {
+ public void afterReplication(IReplicationJob job) {
try {
+ int pendingOps = pendingRepOpsCount.decrementAndGet();
if (job.getOperation() == IReplicationJob.ReplicationOperation.REPLICATE
&& job instanceof ILSMIndexReplicationJob) {
((ILSMIndexReplicationJob) job).endReplication();
}
+ if (pendingOps == 0 && replicationJobsQ.isEmpty()) {
+ closeChannels();
+ }
} catch (HyracksDataException e) {
throw new ReplicationException(e);
}
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationOperation.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationOperation.java
new file mode 100644
index 0000000000..258f24a2de
--- /dev/null
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationOperation.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.replication.management;
+
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.asterix.common.api.INcApplicationContext;
+import org.apache.asterix.common.replication.IPartitionReplica;
+import org.apache.asterix.common.storage.DatasetResourceReference;
+import org.apache.asterix.replication.api.PartitionReplica;
+import org.apache.asterix.replication.api.ReplicationDestination;
+import org.apache.asterix.replication.sync.IndexSynchronizer;
+import org.apache.hyracks.api.replication.IReplicationJob;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
+import org.apache.hyracks.storage.am.lsm.common.impls.AbstractIoOperation;
+import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
+import org.apache.hyracks.storage.am.lsm.common.impls.NoOpIOOperationCallbackFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+public class ReplicationOperation extends AbstractIoOperation {
+
+ private static final Logger LOGGER = LogManager.getLogger();
+
+ private static final ILSMIOOperationCallback INSTANCE =
+ NoOpIOOperationCallbackFactory.INSTANCE.createIoOpCallback(null);
+ private final INcApplicationContext appCtx;
+ private final DatasetResourceReference indexRef;
+ private final IReplicationJob job;
+ private final IndexReplicationManager indexReplicationManager;
+
+ public ReplicationOperation(INcApplicationContext appCtx, DatasetResourceReference indexRef, IReplicationJob job,
+ IndexReplicationManager indexReplicationManager) {
+ super(null, null, INSTANCE, indexRef.getRelativePath().toString());
+ this.appCtx = appCtx;
+ this.indexRef = indexRef;
+ this.job = job;
+ this.indexReplicationManager = indexReplicationManager;
+ }
+
+ @Override
+ public LSMIOOperationType getIOOpertionType() {
+ return LSMIOOperationType.REPLICATE;
+ }
+
+ @Override
+ public LSMIOOperationStatus call() {
+ try {
+ Set<ReplicationDestination> destinations = indexReplicationManager.getDestinations();
+ if (destinations.isEmpty() || indexReplicationManager.skip(indexRef)) {
+ return LSMIOOperationStatus.SUCCESS;
+ }
+ LOGGER.debug("started replicate operation on index {}", indexRef);
+ final IndexSynchronizer synchronizer = new IndexSynchronizer(job, appCtx);
+ final int indexPartition = indexRef.getPartitionId();
+ for (ReplicationDestination dest : destinations) {
+ Optional<IPartitionReplica> partitionReplica = dest.getPartitionReplica(indexPartition);
+ if (partitionReplica.isEmpty()) {
+ continue;
+ }
+ PartitionReplica destReplica = null;
+ try {
+ destReplica = dest.getPartitionReplicaConnection(partitionReplica.get().getIdentifier(), appCtx);
+ synchronizer.sync(destReplica);
+ dest.recycleConnection(destReplica);
+ } catch (Exception e) {
+ if (destReplica != null) {
+ destReplica.close();
+ }
+ indexReplicationManager.handleFailure(dest, e);
+ }
+ }
+ LOGGER.debug("completed replicate operation on index {}", indexRef);
+ return LSMIOOperationStatus.SUCCESS;
+ } finally {
+ indexReplicationManager.afterReplication(job);
+ }
+ }
+
+ @Override
+ protected LSMComponentFileReferences getComponentFiles() {
+ return null;
+ }
+
+ @Override
+ public long getRemainingPages() {
+ return 0;
+ }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractAsynchronousScheduler.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractAsynchronousScheduler.java
index e266a6fc09..049da38fa1 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractAsynchronousScheduler.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractAsynchronousScheduler.java
@@ -38,9 +38,10 @@ public abstract class AbstractAsynchronousScheduler implements ILSMIOOperationSc
private final int maxNumFlushes;
protected final Map<String, ILSMIOOperation> runningFlushOperations = new HashMap<>();
+ protected final Map<String, ILSMIOOperation> runningReplicateOperations = new HashMap<>();
protected final Deque<ILSMIOOperation> waitingFlushOperations = new ArrayDeque<>();
protected final Deque<ILSMIOOperation> waitingMergeOperations = new ArrayDeque<>();
-
+ protected final Deque<ILSMIOOperation> waitingReplicateOperations = new ArrayDeque<>();
protected final Map<String, Throwable> failedGroups = new HashMap<>();
public AbstractAsynchronousScheduler(ThreadFactory threadFactory, final IIoOperationFailedCallback callback,
@@ -58,8 +59,11 @@ public abstract class AbstractAsynchronousScheduler implements ILSMIOOperationSc
case MERGE:
scheduleMerge(operation);
break;
+ case REPLICATE:
+ scheduleReplicate(operation);
+ break;
case NOOP:
- return;
+ break;
default:
// this should never happen
// just guard here to avoid silent failures in case of future extensions
@@ -75,6 +79,10 @@ public abstract class AbstractAsynchronousScheduler implements ILSMIOOperationSc
break;
case MERGE:
completeMerge(operation);
+ break;
+ case REPLICATE:
+ completeReplicate(operation);
+ break;
case NOOP:
return;
default:
@@ -149,6 +157,46 @@ public abstract class AbstractAsynchronousScheduler implements ILSMIOOperationSc
}
}
+ private void scheduleReplicate(ILSMIOOperation operation) {
+ String id = operation.getIndexIdentifier();
+ synchronized (executor) {
+ if (runningReplicateOperations.size() >= maxNumFlushes || runningReplicateOperations.containsKey(id)) {
+ waitingReplicateOperations.add(operation);
+ } else {
+ runningReplicateOperations.put(id, operation);
+ executor.submit(operation);
+ }
+ }
+ }
+
+ private void completeReplicate(ILSMIOOperation operation) {
+ String id = operation.getIndexIdentifier();
+ synchronized (executor) {
+ runningReplicateOperations.remove(id);
+ // Schedule replicate in FIFO order. Must make sure that there is at most one scheduled replicate for each index.
+ for (ILSMIOOperation replicateOp : waitingReplicateOperations) {
+ String replicateOpId = replicateOp.getIndexIdentifier();
+ if (runningReplicateOperations.size() < maxNumFlushes) {
+ if (!runningReplicateOperations.containsKey(replicateOpId) && !replicateOp.isCompleted()) {
+ runningReplicateOperations.put(replicateOpId, replicateOp);
+ executor.submit(replicateOp);
+ }
+ } else {
+ break;
+ }
+ }
+ // cleanup scheduled replicate
+ while (!waitingReplicateOperations.isEmpty()) {
+ ILSMIOOperation top = waitingReplicateOperations.peek();
+ if (top.isCompleted() || runningReplicateOperations.get(top.getIndexIdentifier()) == top) {
+ waitingReplicateOperations.poll();
+ } else {
+ break;
+ }
+ }
+ }
+ }
+
@Override
public void close() throws IOException {
executor.shutdown();
[asterixdb] 10/20: [NO ISSUE][OTH] Fix log message
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 99122833998ccd56d57741efc5af268faa3149ce
Author: Ali Alsuliman <al...@gmail.com>
AuthorDate: Thu Jul 13 22:25:26 2023 -0700
[NO ISSUE][OTH] Fix log message
Change-Id: I7cfd6a0d0ac0cc5073abe0cf634abc59ed627b45
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17641
Reviewed-by: Michael Blow <mb...@apache.org>
Tested-by: Michael Blow <mb...@apache.org>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
---
.../org/apache/asterix/app/message/ExecuteStatementRequestMessage.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
index 9b722df12c..05bc87b2f5 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
@@ -244,6 +244,6 @@ public class ExecuteStatementRequestMessage implements ICcAddressedMessage {
@Override
public String toString() {
return String.format("%s(id=%s, from=%s, uuid=%s): %s", getClass().getSimpleName(), requestMessageId,
- requestNodeId, LogRedactionUtil.statement(statementsText), requestReference.getUuid());
+ requestNodeId, requestReference.getUuid(), LogRedactionUtil.statement(statementsText));
}
}
[asterixdb] 11/20: [NO ISSUE][OTH] Move page size storage config to common config
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit c284d360d3298312dbcaca922e18e0ed1eca8ce5
Author: Murtadha Hubail <mh...@apache.org>
AuthorDate: Mon May 15 19:09:50 2023 +0300
[NO ISSUE][OTH] Move page size storage config to common config
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
- Move page size storage config used during compilation to
the properties common section to avoid warning of usage
outside NC scope.
(cherry picked from commit 33751783)
Change-Id: I43d603feec63261ec14b723d0d39e4a71461750f
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17569
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Michael Blow <mb...@apache.org>
---
.../src/main/resources/{cc.conf => cc-cloud-storage.conf} | 6 +++++-
asterixdb/asterix-app/src/main/resources/cc.conf | 2 +-
asterixdb/asterix-app/src/main/resources/cc2.conf | 2 +-
asterixdb/asterix-app/src/main/resources/cc3.conf | 2 +-
asterixdb/asterix-app/src/main/resources/cc4.conf | 2 +-
asterixdb/asterix-app/src/main/resources/cc_no_cbo.conf | 2 +-
.../resources/{cc.conf => cc_static_partitioning.conf} | 6 +++++-
.../src/test/resources/{cc.conf => cc-analyze.conf} | 14 +++++---------
.../src/test/resources/{cc.conf => cc-cloud-storage.conf} | 6 +++++-
asterixdb/asterix-app/src/test/resources/cc-multipart.conf | 2 +-
.../src/test/resources/{cc.conf => cc-single-lookup.conf} | 6 +++++-
asterixdb/asterix-app/src/test/resources/cc-single.conf | 2 +-
.../src/test/resources/cc-small-txn-log-partition.conf | 2 +-
asterixdb/asterix-app/src/test/resources/cc-ssl.conf | 4 ++--
asterixdb/asterix-app/src/test/resources/cc-storage.conf | 2 +-
asterixdb/asterix-app/src/test/resources/cc.conf | 2 +-
.../results/api/cluster_state_1/cluster_state_1.1.regexadm | 1 +
.../cluster_state_1_full/cluster_state_1_full.1.regexadm | 1 +
.../cluster_state_1_less/cluster_state_1_less.1.regexadm | 1 +
.../apache/asterix/common/config/StorageProperties.java | 1 +
20 files changed, 41 insertions(+), 25 deletions(-)
diff --git a/asterixdb/asterix-app/src/main/resources/cc.conf b/asterixdb/asterix-app/src/main/resources/cc-cloud-storage.conf
similarity index 91%
copy from asterixdb/asterix-app/src/main/resources/cc.conf
copy to asterixdb/asterix-app/src/main/resources/cc-cloud-storage.conf
index dc6e5a2075..a61ac87783 100644
--- a/asterixdb/asterix-app/src/main/resources/cc.conf
+++ b/asterixdb/asterix-app/src/main/resources/cc-cloud-storage.conf
@@ -1,3 +1,5 @@
+<<<<<<< HEAD (5f076a [ASTERIXDB-3221][REPL] Use IO scheduler for replication ops)
+=======
; Licensed to the Apache Software Foundation (ASF) under one
; or more contributor license agreements. See the NOTICE file
; distributed with this work for additional information
@@ -37,7 +39,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
storage.io.scheduler=greedy
@@ -65,3 +66,6 @@ compiler.sort.parallel=false
compiler.internal.sanitycheck=true
messaging.frame.size=4096
messaging.frame.count=512
+cloud.deployment=true
+storage.buffercache.pagesize=32KB
+>>>>>>> CHANGE (337517 [NO ISSUE][OTH] Move page size storage config to common conf)
diff --git a/asterixdb/asterix-app/src/main/resources/cc.conf b/asterixdb/asterix-app/src/main/resources/cc.conf
index dc6e5a2075..b531940447 100644
--- a/asterixdb/asterix-app/src/main/resources/cc.conf
+++ b/asterixdb/asterix-app/src/main/resources/cc.conf
@@ -37,7 +37,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
storage.io.scheduler=greedy
@@ -65,3 +64,4 @@ compiler.sort.parallel=false
compiler.internal.sanitycheck=true
messaging.frame.size=4096
messaging.frame.count=512
+storage.buffercache.pagesize=32KB
diff --git a/asterixdb/asterix-app/src/main/resources/cc2.conf b/asterixdb/asterix-app/src/main/resources/cc2.conf
index c746198745..3f456cfa7d 100644
--- a/asterixdb/asterix-app/src/main/resources/cc2.conf
+++ b/asterixdb/asterix-app/src/main/resources/cc2.conf
@@ -35,7 +35,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
@@ -60,3 +59,4 @@ compiler.windowmemory=192KB
compiler.parallelism=-1
messaging.frame.size=4096
messaging.frame.count=512
+storage.buffercache.pagesize=32KB
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/main/resources/cc3.conf b/asterixdb/asterix-app/src/main/resources/cc3.conf
index 8e9b5ab407..aac7cbe33b 100644
--- a/asterixdb/asterix-app/src/main/resources/cc3.conf
+++ b/asterixdb/asterix-app/src/main/resources/cc3.conf
@@ -35,7 +35,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
@@ -60,3 +59,4 @@ compiler.windowmemory=192KB
compiler.parallelism=3
messaging.frame.size=4096
messaging.frame.count=512
+storage.buffercache.pagesize=32KB
diff --git a/asterixdb/asterix-app/src/main/resources/cc4.conf b/asterixdb/asterix-app/src/main/resources/cc4.conf
index 6a66d250ac..bf37499874 100644
--- a/asterixdb/asterix-app/src/main/resources/cc4.conf
+++ b/asterixdb/asterix-app/src/main/resources/cc4.conf
@@ -35,7 +35,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
[cc]
address = 127.0.0.1
@@ -54,3 +53,4 @@ compiler.windowmemory=192KB
messaging.frame.size=4096
messaging.frame.count=512
compiler.parallelism=-1
+storage.buffercache.pagesize=32KB
diff --git a/asterixdb/asterix-app/src/main/resources/cc_no_cbo.conf b/asterixdb/asterix-app/src/main/resources/cc_no_cbo.conf
index deb92e0bb2..ab028c859b 100644
--- a/asterixdb/asterix-app/src/main/resources/cc_no_cbo.conf
+++ b/asterixdb/asterix-app/src/main/resources/cc_no_cbo.conf
@@ -37,7 +37,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
storage.io.scheduler=greedy
@@ -64,3 +63,4 @@ compiler.sort.parallel=false
compiler.internal.sanitycheck=true
messaging.frame.size=4096
messaging.frame.count=512
+storage.buffercache.pagesize=32KB
diff --git a/asterixdb/asterix-app/src/main/resources/cc.conf b/asterixdb/asterix-app/src/main/resources/cc_static_partitioning.conf
similarity index 91%
copy from asterixdb/asterix-app/src/main/resources/cc.conf
copy to asterixdb/asterix-app/src/main/resources/cc_static_partitioning.conf
index dc6e5a2075..e3e779e17d 100644
--- a/asterixdb/asterix-app/src/main/resources/cc.conf
+++ b/asterixdb/asterix-app/src/main/resources/cc_static_partitioning.conf
@@ -1,3 +1,5 @@
+<<<<<<< HEAD (5f076a [ASTERIXDB-3221][REPL] Use IO scheduler for replication ops)
+=======
; Licensed to the Apache Software Foundation (ASF) under one
; or more contributor license agreements. See the NOTICE file
; distributed with this work for additional information
@@ -37,7 +39,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
storage.io.scheduler=greedy
@@ -65,3 +66,6 @@ compiler.sort.parallel=false
compiler.internal.sanitycheck=true
messaging.frame.size=4096
messaging.frame.count=512
+storage.buffercache.pagesize=32KB
+storage.partitioning=static
+>>>>>>> CHANGE (337517 [NO ISSUE][OTH] Move page size storage config to common conf)
diff --git a/asterixdb/asterix-app/src/test/resources/cc.conf b/asterixdb/asterix-app/src/test/resources/cc-analyze.conf
similarity index 90%
copy from asterixdb/asterix-app/src/test/resources/cc.conf
copy to asterixdb/asterix-app/src/test/resources/cc-analyze.conf
index 9532849649..d8b35122a1 100644
--- a/asterixdb/asterix-app/src/test/resources/cc.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-analyze.conf
@@ -1,3 +1,5 @@
+<<<<<<< HEAD (5f076a [ASTERIXDB-3221][REPL] Use IO scheduler for replication ops)
+=======
; Licensed to the Apache Software Foundation (ASF) under one
; or more contributor license agreements. See the NOTICE file
; distributed with this work for additional information
@@ -39,7 +41,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
@@ -53,11 +54,6 @@ credential.file=src/test/resources/security/passwd
[common]
log.dir = logs/
log.level = INFO
-compiler.framesize=32KB
-compiler.sortmemory=320KB
-compiler.groupmemory=160KB
-compiler.joinmemory=256KB
-compiler.textsearchmemory=160KB
-compiler.windowmemory=192KB
-messaging.frame.size=4096
-messaging.frame.count=512
+compiler.groupmemory=64MB
+storage.buffercache.pagesize=32KB
+>>>>>>> CHANGE (337517 [NO ISSUE][OTH] Move page size storage config to common conf)
diff --git a/asterixdb/asterix-app/src/test/resources/cc.conf b/asterixdb/asterix-app/src/test/resources/cc-cloud-storage.conf
similarity index 91%
copy from asterixdb/asterix-app/src/test/resources/cc.conf
copy to asterixdb/asterix-app/src/test/resources/cc-cloud-storage.conf
index 9532849649..18627b4e83 100644
--- a/asterixdb/asterix-app/src/test/resources/cc.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-cloud-storage.conf
@@ -1,3 +1,5 @@
+<<<<<<< HEAD (5f076a [ASTERIXDB-3221][REPL] Use IO scheduler for replication ops)
+=======
; Licensed to the Apache Software Foundation (ASF) under one
; or more contributor license agreements. See the NOTICE file
; distributed with this work for additional information
@@ -39,7 +41,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
@@ -61,3 +62,6 @@ compiler.textsearchmemory=160KB
compiler.windowmemory=192KB
messaging.frame.size=4096
messaging.frame.count=512
+cloud.deployment=true
+storage.buffercache.pagesize=32KB
+>>>>>>> CHANGE (337517 [NO ISSUE][OTH] Move page size storage config to common conf)
diff --git a/asterixdb/asterix-app/src/test/resources/cc-multipart.conf b/asterixdb/asterix-app/src/test/resources/cc-multipart.conf
index 5543e512e8..6c676246fd 100644
--- a/asterixdb/asterix-app/src/test/resources/cc-multipart.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-multipart.conf
@@ -35,7 +35,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
@@ -54,3 +53,4 @@ compiler.joinmemory=256KB
compiler.windowmemory=192KB
messaging.frame.size=4096
messaging.frame.count=512
+storage.buffercache.pagesize=32KB
diff --git a/asterixdb/asterix-app/src/test/resources/cc.conf b/asterixdb/asterix-app/src/test/resources/cc-single-lookup.conf
similarity index 91%
copy from asterixdb/asterix-app/src/test/resources/cc.conf
copy to asterixdb/asterix-app/src/test/resources/cc-single-lookup.conf
index 9532849649..1b006b3180 100644
--- a/asterixdb/asterix-app/src/test/resources/cc.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-single-lookup.conf
@@ -1,3 +1,5 @@
+<<<<<<< HEAD (5f076a [ASTERIXDB-3221][REPL] Use IO scheduler for replication ops)
+=======
; Licensed to the Apache Software Foundation (ASF) under one
; or more contributor license agreements. See the NOTICE file
; distributed with this work for additional information
@@ -39,7 +41,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
@@ -59,5 +60,8 @@ compiler.groupmemory=160KB
compiler.joinmemory=256KB
compiler.textsearchmemory=160KB
compiler.windowmemory=192KB
+compiler.batch.lookup=false
messaging.frame.size=4096
messaging.frame.count=512
+storage.buffercache.pagesize=32KB
+>>>>>>> CHANGE (337517 [NO ISSUE][OTH] Move page size storage config to common conf)
diff --git a/asterixdb/asterix-app/src/test/resources/cc-single.conf b/asterixdb/asterix-app/src/test/resources/cc-single.conf
index 794b5b0ab9..a6f4e8999a 100644
--- a/asterixdb/asterix-app/src/test/resources/cc-single.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-single.conf
@@ -27,7 +27,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
@@ -48,3 +47,4 @@ compiler.textsearchmemory=160KB
compiler.windowmemory=192KB
messaging.frame.size=4096
messaging.frame.count=512
+storage.buffercache.pagesize=32KB
diff --git a/asterixdb/asterix-app/src/test/resources/cc-small-txn-log-partition.conf b/asterixdb/asterix-app/src/test/resources/cc-small-txn-log-partition.conf
index dd8edb44f4..ae91f5e10d 100644
--- a/asterixdb/asterix-app/src/test/resources/cc-small-txn-log-partition.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-small-txn-log-partition.conf
@@ -33,7 +33,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=128MB
@@ -56,3 +55,4 @@ txn.log.partitionsize=2MB
txn.log.buffer.pagesize=128KB
txn.log.checkpoint.pollfrequency=2147483647
txn.log.checkpoint.history=0
+storage.buffercache.pagesize=32KB
diff --git a/asterixdb/asterix-app/src/test/resources/cc-ssl.conf b/asterixdb/asterix-app/src/test/resources/cc-ssl.conf
index ff7e73b8b2..1498244578 100644
--- a/asterixdb/asterix-app/src/test/resources/cc-ssl.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-ssl.conf
@@ -43,7 +43,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
@@ -69,4 +68,5 @@ compiler.textsearchmemory=160KB
compiler.windowmemory=192KB
messaging.frame.size=4096
messaging.frame.count=512
-ssl.enabled=true
\ No newline at end of file
+ssl.enabled=true
+storage.buffercache.pagesize=32KB
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/cc-storage.conf b/asterixdb/asterix-app/src/test/resources/cc-storage.conf
index 9e869c7bf3..68faa17905 100644
--- a/asterixdb/asterix-app/src/test/resources/cc-storage.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-storage.conf
@@ -33,7 +33,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
@@ -53,6 +52,7 @@ compiler.textsearchmemory=160KB
compiler.windowmemory=192KB
messaging.frame.size=4096
messaging.frame.count=512
+storage.buffercache.pagesize=32KB
storage.compression.block=snappy
txn.log.checkpoint.pollfrequency=10
txn.dataset.checkpoint.interval=10
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/cc.conf b/asterixdb/asterix-app/src/test/resources/cc.conf
index 9532849649..624455c594 100644
--- a/asterixdb/asterix-app/src/test/resources/cc.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc.conf
@@ -39,7 +39,6 @@ address=127.0.0.1
command=asterixnc
app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
-storage.buffercache.pagesize=32KB
storage.buffercache.size=128MB
storage.memorycomponent.globalbudget=512MB
@@ -61,3 +60,4 @@ compiler.textsearchmemory=160KB
compiler.windowmemory=192KB
messaging.frame.size=4096
messaging.frame.count=512
+storage.buffercache.pagesize=32KB
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1/cluster_state_1.1.regexadm b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1/cluster_state_1.1.regexadm
index dd6ac17d84..cb10dd2276 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1/cluster_state_1.1.regexadm
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1/cluster_state_1.1.regexadm
@@ -48,6 +48,7 @@
"replication\.strategy" : "none",
"replication\.timeout" : 120,
"ssl\.enabled" : false,
+ "storage.buffercache.pagesize" : 32768,
"storage.compression.block" : "snappy",
"storage.global.cleanup.timeout" : 600,
"storage.lsm.bloomfilter.falsepositiverate" : 0.01,
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_full/cluster_state_1_full.1.regexadm b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_full/cluster_state_1_full.1.regexadm
index 9103663bfd..59d595bffb 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_full/cluster_state_1_full.1.regexadm
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_full/cluster_state_1_full.1.regexadm
@@ -48,6 +48,7 @@
"replication\.strategy" : "none",
"replication\.timeout" : 120,
"ssl\.enabled" : false,
+ "storage.buffercache.pagesize" : 32768,
"storage.compression.block" : "snappy",
"storage.global.cleanup.timeout" : 600,
"storage.lsm.bloomfilter.falsepositiverate" : 0.01,
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_less/cluster_state_1_less.1.regexadm b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_less/cluster_state_1_less.1.regexadm
index 7ae9886fb5..56b413d190 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_less/cluster_state_1_less.1.regexadm
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_less/cluster_state_1_less.1.regexadm
@@ -48,6 +48,7 @@
"replication\.strategy" : "none",
"replication\.timeout" : 120,
"ssl\.enabled" : false,
+ "storage.buffercache.pagesize" : 32768,
"storage.compression.block" : "snappy",
"storage.global.cleanup.timeout" : 600,
"storage.lsm.bloomfilter.falsepositiverate" : 0.01,
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/StorageProperties.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/StorageProperties.java
index 12c9c684f4..6264f64cf4 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/StorageProperties.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/StorageProperties.java
@@ -74,6 +74,7 @@ public class StorageProperties extends AbstractProperties {
@Override
public Section section() {
switch (this) {
+ case STORAGE_BUFFERCACHE_PAGESIZE:
case STORAGE_COMPRESSION_BLOCK:
case STORAGE_LSM_BLOOMFILTER_FALSEPOSITIVERATE:
case STORAGE_GLOBAL_CLEANUP:
[asterixdb] 14/20: Merge branch 'gerrit/stabilization-667a908755' into neo
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 04b1dd4f4100fac223a9000c069b9abadf353f6e
Merge: c284d360d3 e64d046289
Author: Murtadha Hubail <mh...@apache.org>
AuthorDate: Wed Jul 19 02:56:45 2023 +0300
Merge branch 'gerrit/stabilization-667a908755' into neo
Change-Id: Id1d7944ff31349db441985e7f058e59faa934084
.../asterix/common/context/PrimaryIndexOperationTracker.java | 7 +++----
1 file changed, 3 insertions(+), 4 deletions(-)
[asterixdb] 15/20: Merge branch 'gerrit/mad-hatter' into 'gerrit/cheshire-cat'
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 1e5910e2df64d1527d1a776c273451cc83254f8b
Merge: 35c42acd3c 84c47cf88a
Author: Michael Blow <mi...@couchbase.com>
AuthorDate: Wed Jul 19 09:10:05 2023 -0400
Merge branch 'gerrit/mad-hatter' into 'gerrit/cheshire-cat'
Change-Id: Ibc4d62ded50996c714419d90dadb833c5866dec0
.../appended-resources/supplemental-models.xml | 90 +++++++++++-----------
....com_netty_netty_netty-4.1.94.Final_NOTICE.txt} | 0
hyracks-fullstack/pom.xml | 2 +-
3 files changed, 46 insertions(+), 46 deletions(-)
diff --cc asterixdb/src/main/appended-resources/supplemental-models.xml
index 2300c925d0,5e871f2bb0..6a0f8d0884
--- a/asterixdb/src/main/appended-resources/supplemental-models.xml
+++ b/asterixdb/src/main/appended-resources/supplemental-models.xml
@@@ -160,9 -160,9 +160,9 @@@
<artifactId>netty-transport</artifactId>
<properties>
<!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
-- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
-- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
-- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
</properties>
</project>
</supplement>
@@@ -172,9 -172,9 +172,9 @@@
<artifactId>netty-transport-classes-epoll</artifactId>
<properties>
<!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
-- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
-- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
-- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
</properties>
</project>
</supplement>
@@@ -184,9 -184,9 +184,9 @@@
<artifactId>netty-transport-native-unix-common</artifactId>
<properties>
<!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
-- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
-- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
-- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
</properties>
</project>
</supplement>
@@@ -196,45 -196,9 +196,45 @@@
<artifactId>netty-codec</artifactId>
<properties>
<!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
-- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
-- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
-- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
+ </properties>
+ </project>
+ </supplement>
+ <supplement>
+ <project>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-codec-dns</artifactId>
+ <properties>
+ <!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
+ </properties>
+ </project>
+ </supplement>
+ <supplement>
+ <project>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-codec-http2</artifactId>
+ <properties>
+ <!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
+ </properties>
+ </project>
+ </supplement>
+ <supplement>
+ <project>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-codec-socks</artifactId>
+ <properties>
+ <!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
</properties>
</project>
</supplement>
@@@ -244,21 -208,9 +244,21 @@@
<artifactId>netty-handler</artifactId>
<properties>
<!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
-- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
-- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
-- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
+ </properties>
+ </project>
+ </supplement>
+ <supplement>
+ <project>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-handler-proxy</artifactId>
+ <properties>
+ <!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
</properties>
</project>
</supplement>
@@@ -268,9 -220,9 +268,9 @@@
<artifactId>netty-buffer</artifactId>
<properties>
<!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
-- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
-- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
-- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
</properties>
</project>
</supplement>
@@@ -280,9 -232,9 +280,9 @@@
<artifactId>netty-common</artifactId>
<properties>
<!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
-- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
-- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
-- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
</properties>
</project>
</supplement>
@@@ -292,9 -244,9 +292,9 @@@
<artifactId>netty-codec-http</artifactId>
<properties>
<!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
-- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
-- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
-- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
</properties>
</project>
</supplement>
@@@ -304,21 -256,9 +304,21 @@@
<artifactId>netty-resolver</artifactId>
<properties>
<!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
-- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
-- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
-- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
+ </properties>
+ </project>
+ </supplement>
+ <supplement>
+ <project>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-resolver-dns</artifactId>
+ <properties>
+ <!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
</properties>
</project>
</supplement>
@@@ -328,9 -268,9 +328,9 @@@
<artifactId>netty-transport-native-unix-common</artifactId>
<properties>
<!-- netty is ALv2, and does not contain any embedded LICENSE or NOTICE file -->
-- <license.ignoreMissingEmbeddedLicense>4.1.87.Final</license.ignoreMissingEmbeddedLicense>
-- <license.ignoreMissingEmbeddedNotice>4.1.87.Final</license.ignoreMissingEmbeddedNotice>
-- <license.ignoreNoticeOverride>4.1.87.Final</license.ignoreNoticeOverride>
++ <license.ignoreMissingEmbeddedLicense>4.1.94.Final</license.ignoreMissingEmbeddedLicense>
++ <license.ignoreMissingEmbeddedNotice>4.1.94.Final</license.ignoreMissingEmbeddedNotice>
++ <license.ignoreNoticeOverride>4.1.94.Final</license.ignoreNoticeOverride>
</properties>
</project>
</supplement>
diff --cc hyracks-fullstack/pom.xml
index 9380fb181d,8311c5cfad..76f7b10441
--- a/hyracks-fullstack/pom.xml
+++ b/hyracks-fullstack/pom.xml
@@@ -76,7 -76,6 +76,7 @@@
<snappy.version>1.1.8.4</snappy.version>
<jackson.version>2.14.1</jackson.version>
<jackson-databind.version>${jackson.version}</jackson-databind.version>
- <netty.version>4.1.87.Final</netty.version>
++ <netty.version>4.1.94.Final</netty.version>
<implementation.title>Apache Hyracks and Algebricks - ${project.name}</implementation.title>
<implementation.url>https://asterixdb.apache.org/</implementation.url>
[asterixdb] 05/20: [ASTERIXDB-3219][REPL] Add timeout to log replication
Posted by mb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mblow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 2f56ed9d6d394fe990e83f96b9d8683fabfc1064
Author: Murtadha Hubail <mh...@apache.org>
AuthorDate: Fri Jul 7 16:26:40 2023 -0700
[ASTERIXDB-3219][REPL] Add timeout to log replication
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
- Do not wait indefinitely for logs to be replicated.
Change-Id: I53b3a0d23514fce09082556e031f822dfe426a35
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17632
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Ali Alsuliman <al...@gmail.com>
---
.../service/logging/LogManagerWithReplication.java | 19 ++++++++++++++++++-
1 file changed, 18 insertions(+), 1 deletion(-)
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManagerWithReplication.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManagerWithReplication.java
index 8a1cc656fa..1c614c902f 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManagerWithReplication.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManagerWithReplication.java
@@ -18,6 +18,8 @@
*/
package org.apache.asterix.transaction.management.service.logging;
+import java.util.concurrent.TimeUnit;
+
import org.apache.asterix.common.exceptions.ACIDException;
import org.apache.asterix.common.replication.IReplicationManager;
import org.apache.asterix.common.replication.IReplicationStrategy;
@@ -26,6 +28,7 @@ import org.apache.asterix.common.transactions.ITransactionSubsystem;
import org.apache.asterix.common.transactions.LogSource;
import org.apache.asterix.common.transactions.LogType;
import org.apache.hyracks.api.util.InvokeUtil;
+import org.apache.logging.log4j.Logger;
import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
import it.unimi.dsi.fastutil.longs.LongSet;
@@ -33,12 +36,16 @@ import it.unimi.dsi.fastutil.longs.LongSets;
public class LogManagerWithReplication extends LogManager {
+ private static final Logger LOGGER = org.apache.logging.log4j.LogManager.getLogger();
private IReplicationManager replicationManager;
private IReplicationStrategy replicationStrategy;
private final LongSet replicatedTxn = LongSets.synchronize(new LongOpenHashSet());
+ private final long replicationTimeoutMillis;
public LogManagerWithReplication(ITransactionSubsystem txnSubsystem) {
super(txnSubsystem);
+ replicationTimeoutMillis = TimeUnit.SECONDS
+ .toMillis(txnSubsystem.getApplicationContext().getReplicationProperties().getReplicationTimeOut());
}
@SuppressWarnings("squid:S2445")
@@ -94,8 +101,18 @@ public class LogManagerWithReplication extends LogManager {
//wait for job Commit/Abort ACK from replicas
if (logRecord.isReplicate() && (logRecord.getLogType() == LogType.JOB_COMMIT
|| logRecord.getLogType() == LogType.ABORT)) {
+ long replicationTimeOut = replicationTimeoutMillis;
while (!logRecord.isReplicated()) {
- logRecord.wait();
+ if (replicationTimeOut <= 0) {
+ LOGGER.warn(
+ "{} ms passed without receiving acks for log {}; setting log as replicated due to timeout",
+ replicationTimeoutMillis, logRecord.getLogRecordForDisplay());
+ logRecord.setReplicated(true);
+ continue;
+ }
+ final long startTime = System.nanoTime();
+ logRecord.wait(replicationTimeOut);
+ replicationTimeOut -= TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime);
}
}
}