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 2026/01/16 15:58:21 UTC
(asterixdb) branch master updated (ef1dac546a -> 91872ef6b5)
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 ef1dac546a Merge branch 'gerrit/morpheus' into 'master'
new d94d002325 [NO ISSUE] Update ch2 data path
new 9044ddcefa [ASTERIXDB-3689][HYR][NET] MuxDemux negotiation hardening
new 91872ef6b5 Merge branch 'gerrit/phoenix' into 'master'
The 3 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:
(asterixdb) 03/03: Merge branch 'gerrit/phoenix' 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 91872ef6b5aed6e4c4fcebe966e2a4f35692a1eb
Merge: ef1dac546a 9044ddcefa
Author: Michael Blow <mi...@couchbase.com>
AuthorDate: Fri Jan 16 10:56:09 2026 -0500
Merge branch 'gerrit/phoenix' into 'master'
* [NO ISSUE] Update ch2 data path (MB-58891)
* [ASTERIXDB-3689][HYR][NET] MuxDemux negotiation hardening (MB-70113,MB-70114)
Change-Id: I93d18019a71d594328705627cc02f76a40963a2d
(asterixdb) 02/03: [ASTERIXDB-3689][HYR][NET] MuxDemux negotiation hardening
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 9044ddcefaf8eb7e1f88d91cc11a627416d4bd41
Author: Michael Blow <mi...@couchbase.com>
AuthorDate: Wed Jan 14 10:35:13 2026 -0500
[ASTERIXDB-3689][HYR][NET] MuxDemux negotiation hardening
(cherry picked from commit 0646f96bf6aa)
Ext-ref: MB-70113,MB-70114
Change-Id: I3887c992ad73f501f074ff5756f6204506043d65
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/20797
Tested-by: Michael Blow <mb...@apache.org>
Reviewed-by: Michael Blow <mb...@apache.org>
---
.../hyracks/net/protocols/muxdemux/MuxDemux.java | 42 +++++++++---
.../hyracks/net/protocols/tcp/TCPEndpoint.java | 74 ++++++++++++----------
2 files changed, 73 insertions(+), 43 deletions(-)
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java
index 3189ea11d2..0837b294b0 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java
@@ -30,6 +30,8 @@ import org.apache.hyracks.net.protocols.tcp.ITCPConnectionListener;
import org.apache.hyracks.net.protocols.tcp.TCPConnection;
import org.apache.hyracks.net.protocols.tcp.TCPEndpoint;
import org.apache.hyracks.util.JSONUtil;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
@@ -43,6 +45,8 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
* @author vinayakb
*/
public class MuxDemux {
+ private static final Logger LOGGER = LogManager.getLogger();
+
private final InetSocketAddress localAddress;
private final IChannelOpenListener channelOpenListener;
@@ -56,7 +60,7 @@ public class MuxDemux {
private final MuxDemuxPerformanceCounters perfCounters;
- private final IChannelInterfaceFactory channelInterfaceFatory;
+ private final IChannelInterfaceFactory channelInterfaceFactory;
/**
* Constructor.
@@ -80,7 +84,7 @@ public class MuxDemux {
this.localAddress = localAddress;
this.channelOpenListener = listener;
this.maxConnectionAttempts = maxConnectionAttempts;
- this.channelInterfaceFatory = channelInterfaceFactory;
+ this.channelInterfaceFactory = channelInterfaceFactory;
outgoingConnectionMap = new HashMap<>();
incomingConnectionMap = new HashMap<>();
this.tcpEndpoint = new TCPEndpoint(new ITCPConnectionListener() {
@@ -90,7 +94,10 @@ public class MuxDemux {
synchronized (MuxDemux.this) {
mConn = outgoingConnectionMap.get(connection.getRemoteAddress());
}
- assert mConn != null;
+ if (mConn == null) {
+ throw new IllegalStateException(
+ "connectionEstablished called for unknown connection: " + connection.getRemoteAddress());
+ }
mConn.setTCPConnection(connection);
connection.setEventListener(mConn);
connection.setAttachment(mConn);
@@ -110,10 +117,19 @@ public class MuxDemux {
MultiplexedConnection mConn;
synchronized (MuxDemux.this) {
mConn = outgoingConnectionMap.get(remoteAddress);
- assert mConn != null;
+ if (mConn == null) {
+ throw new IllegalStateException(
+ "connectionFailure called for unknown connection: " + remoteAddress);
+ }
int nConnectionAttempts = mConn.getConnectionAttempts();
if (nConnectionAttempts > MuxDemux.this.maxConnectionAttempts) {
- outgoingConnectionMap.remove(remoteAddress);
+ if (outgoingConnectionMap.remove(remoteAddress) == null) {
+ LOGGER.warn("connection to {} failed after {} attempts, but connection was already removed",
+ remoteAddress, nConnectionAttempts);
+ } else {
+ LOGGER.debug("connection to {} failed after {} attempts, giving up", remoteAddress,
+ nConnectionAttempts);
+ }
mConn.setConnectionFailure(new IOException(remoteAddress.toString() + ": " + error, error));
} else {
mConn.setConnectionAttempts(nConnectionAttempts + 1);
@@ -126,9 +142,19 @@ public class MuxDemux {
public void connectionClosed(TCPConnection connection) {
synchronized (MuxDemux.this) {
if (connection.getType() == TCPConnection.ConnectionType.OUTGOING) {
- outgoingConnectionMap.remove(connection.getRemoteAddress());
+ if (outgoingConnectionMap.remove(connection.getRemoteAddress()) == null) {
+ LOGGER.warn("outgoing connection to {} already removed on close",
+ connection.getRemoteAddress());
+ } else {
+ LOGGER.debug("removed outgoing connection to {} on close", connection.getRemoteAddress());
+ }
} else if (connection.getType() == TCPConnection.ConnectionType.INCOMING) {
- incomingConnectionMap.remove(connection.getRemoteAddress());
+ if (incomingConnectionMap.remove(connection.getRemoteAddress()) == null) {
+ LOGGER.warn("incoming connection from {} already removed on close",
+ connection.getRemoteAddress());
+ } else {
+ LOGGER.debug("removed incoming connection from {} on close", connection.getRemoteAddress());
+ }
}
}
}
@@ -200,7 +226,7 @@ public class MuxDemux {
* @return
*/
public IChannelInterfaceFactory getChannelInterfaceFactory() {
- return channelInterfaceFatory;
+ return channelInterfaceFactory;
}
public synchronized JsonNode getState() {
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/tcp/TCPEndpoint.java b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/tcp/TCPEndpoint.java
index b75e3c6363..d6ffe23e81 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/tcp/TCPEndpoint.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/tcp/TCPEndpoint.java
@@ -110,7 +110,7 @@ public class TCPEndpoint {
@Override
public String toString() {
- return "TCPEndpoint [Local Address: " + localAddress + "]";
+ return "TCPEndpoint [" + (localAddress == null ? "<outgoing>" : "Local Address: " + localAddress) + "]";
}
private class IOThread extends Thread {
@@ -127,7 +127,7 @@ public class TCPEndpoint {
private final Selector selector;
public IOThread() throws IOException {
- super("TCPEndpoint IO Thread [" + localAddress + "]");
+ super("TCPEndpoint IO Thread [" + (localAddress == null ? "<outgoing>" : localAddress) + "]");
setDaemon(true);
setPriority(Thread.NORM_PRIORITY);
this.pendingConnections = new ArrayList<>();
@@ -144,44 +144,44 @@ public class TCPEndpoint {
try {
int n = selector.select();
collectOutstandingWork();
- if (!workingPendingConnections.isEmpty()) {
- for (InetSocketAddress address : workingPendingConnections) {
- SocketChannel channel = SocketChannel.open();
- register(channel);
- boolean connect = false;
- boolean failure = false;
- try {
- connect = channel.connect(address);
- } catch (IOException e) {
- failure = true;
- synchronized (connectionListener) {
- connectionListener.connectionFailure(address, e);
- }
+ for (Iterator<InetSocketAddress> iterator = workingPendingConnections.iterator(); iterator
+ .hasNext();) {
+ InetSocketAddress address = iterator.next();
+ iterator.remove();
+ SocketChannel channel = SocketChannel.open();
+ register(channel);
+ boolean connect = false;
+ boolean failure = false;
+ try {
+ connect = channel.connect(address);
+ } catch (IOException e) {
+ failure = true;
+ synchronized (connectionListener) {
+ connectionListener.connectionFailure(address, e);
}
- if (!failure) {
- if (!connect) {
- SelectionKey key = channel.register(selector, SelectionKey.OP_CONNECT);
- key.attach(address);
- } else {
- socketConnected(address, channel);
- }
+ }
+ if (!failure) {
+ if (!connect) {
+ SelectionKey key = channel.register(selector, SelectionKey.OP_CONNECT);
+ key.attach(address);
+ } else {
+ socketConnected(address, channel);
}
}
- workingPendingConnections.clear();
}
- if (!workingIncomingConnections.isEmpty()) {
- for (SocketChannel channel : workingIncomingConnections) {
- register(channel);
- connectionReceived(channel);
- }
- workingIncomingConnections.clear();
+ for (Iterator<SocketChannel> iterator = workingIncomingConnections.iterator(); iterator
+ .hasNext();) {
+ SocketChannel channel = iterator.next();
+ iterator.remove();
+ register(channel);
+ connectionReceived(channel);
}
synchronized (handshakeCompletedConnections) {
- if (!handshakeCompletedConnections.isEmpty()) {
- for (final PendingHandshakeConnection conn : handshakeCompletedConnections) {
- handshakeCompleted(conn);
- }
- handshakeCompletedConnections.clear();
+ for (Iterator<PendingHandshakeConnection> iterator =
+ handshakeCompletedConnections.iterator(); iterator.hasNext();) {
+ PendingHandshakeConnection conn = iterator.next();
+ iterator.remove();
+ handshakeCompleted(conn);
}
}
if (n > 0) {
@@ -257,7 +257,11 @@ public class TCPEndpoint {
}
} catch (Exception e) {
LOGGER.error("failed to establish connection after handshake", e);
- handleHandshakeFailure(conn);
+ try {
+ handleHandshakeFailure(conn);
+ } catch (Exception ex) {
+ e.addSuppressed(ex);
+ }
}
}
(asterixdb) 01/03: [NO ISSUE] Update ch2 data path
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 d94d002325644d3b5996bee689a3c7028c679120
Author: Ian Maxon <im...@apache.org>
AuthorDate: Tue Jan 6 11:00:08 2026 -0800
[NO ISSUE] Update ch2 data path
Details:
- The path where the ch2 data generator stores tables
was updated upstream. Simply adjust the DDL file to
use this new path.
(cherry picked from commit 38ae445bef)
Ext-ref: MB-58891
Change-Id: If77144159c351dab051a6aebc2a2dea779cd8c10
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/20791
Reviewed-by: Ian Maxon <im...@apache.org>
Reviewed-by: Michael Blow <mb...@apache.org>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
---
asterixdb/NOTICE | 2 +-
.../queries_sqlpp/cbo-join/ch2/ch2.2.update.sqlpp | 22 +++++++++++-----------
hyracks-fullstack/NOTICE | 2 +-
3 files changed, 13 insertions(+), 13 deletions(-)
diff --git a/asterixdb/NOTICE b/asterixdb/NOTICE
index 5118782978..a44569bc27 100644
--- a/asterixdb/NOTICE
+++ b/asterixdb/NOTICE
@@ -1,5 +1,5 @@
Apache AsterixDB
-Copyright 2015-2025 The Apache Software Foundation
+Copyright 2015-2026 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/cbo-join/ch2/ch2.2.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/cbo-join/ch2/ch2.2.update.sqlpp
index 6c7181c62c..eb6b7874ce 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/cbo-join/ch2/ch2.2.update.sqlpp
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/cbo-join/ch2/ch2.2.update.sqlpp
@@ -20,24 +20,24 @@ use test;
-LOAD DATASET customer USING localfs ((`path`=`asterix_nc1://target/data/ch2/customer-0-0.json`),(`format`=`json`));
+LOAD DATASET customer USING localfs ((`path`=`asterix_nc1://target/data/ch2/customer/customer-0-0.json`),(`format`=`json`));
-LOAD DATASET district USING localfs ((`path`=`asterix_nc1://target/data/ch2/district-0-0.json`),(`format`=`json`));
+LOAD DATASET district USING localfs ((`path`=`asterix_nc1://target/data/ch2/district/district-0-0.json`),(`format`=`json`));
-LOAD DATASET history USING localfs ((`path`=`asterix_nc1://target/data/ch2/history-0-0.json`),(`format`=`json`));
+LOAD DATASET history USING localfs ((`path`=`asterix_nc1://target/data/ch2/history/history-0-0.json`),(`format`=`json`));
-LOAD DATASET item USING localfs ((`path`=`asterix_nc1://target/data/ch2/item-0-0.json`),(`format`=`json`));
+LOAD DATASET item USING localfs ((`path`=`asterix_nc1://target/data/ch2/item/item-0-0.json`),(`format`=`json`));
-LOAD DATASET nation USING localfs ((`path`=`asterix_nc1://target/data/ch2/nation-0-0.json`),(`format`=`json`));
+LOAD DATASET nation USING localfs ((`path`=`asterix_nc1://target/data/ch2/nation/nation-0-0.json`),(`format`=`json`));
-LOAD DATASET neworder USING localfs ((`path`=`asterix_nc1://target/data/ch2/neworder-0-0.json`),(`format`=`json`));
+LOAD DATASET neworder USING localfs ((`path`=`asterix_nc1://target/data/ch2/neworder/neworder-0-0.json`),(`format`=`json`));
-LOAD DATASET orders USING localfs ((`path`=`asterix_nc1://target/data/ch2/orders-0-0.json`),(`format`=`json`));
+LOAD DATASET orders USING localfs ((`path`=`asterix_nc1://target/data/ch2/orders/orders-0-0.json`),(`format`=`json`));
-LOAD DATASET region USING localfs ((`path`=`asterix_nc1://target/data/ch2/region-0-0.json`),(`format`=`json`));
+LOAD DATASET region USING localfs ((`path`=`asterix_nc1://target/data/ch2/region/region-0-0.json`),(`format`=`json`));
-LOAD DATASET stock USING localfs ((`path`=`asterix_nc1://target/data/ch2/stock-0-0.json`),(`format`=`json`));
+LOAD DATASET stock USING localfs ((`path`=`asterix_nc1://target/data/ch2/stock/stock-0-0.json`),(`format`=`json`));
-LOAD DATASET supplier USING localfs ((`path`=`asterix_nc1://target/data/ch2/supplier-0-0.json`),(`format`=`json`));
+LOAD DATASET supplier USING localfs ((`path`=`asterix_nc1://target/data/ch2/supplier/supplier-0-0.json`),(`format`=`json`));
-LOAD DATASET warehouse USING localfs ((`path`=`asterix_nc1://target/data/ch2/warehouse-0-0.json`),(`format`=`json`));
+LOAD DATASET warehouse USING localfs ((`path`=`asterix_nc1://target/data/ch2/warehouse/warehouse-0-0.json`),(`format`=`json`));
diff --git a/hyracks-fullstack/NOTICE b/hyracks-fullstack/NOTICE
index 722db88282..9050c829f3 100644
--- a/hyracks-fullstack/NOTICE
+++ b/hyracks-fullstack/NOTICE
@@ -1,5 +1,5 @@
Apache Hyracks and Algebricks
-Copyright 2015-2025 The Apache Software Foundation
+Copyright 2015-2026 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).