This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 9.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
The following commit(s) were added to refs/heads/9.0.x by this push:
new c478fa4bf5 Various minor fixes and improvements from code review
c478fa4bf5 is described below
commit c478fa4bf50244464454956b0c8edc6e6d541ab1
Author: remm <[email protected]>
AuthorDate: Wed Jul 8 15:47:30 2026 +0200
Various minor fixes and improvements from code review
hashCode seemed odd in many places (generated from four bytes), probably
trying to optimize it. Replace with a more traditional way. If there is
an actual performance impact, I will revert it.
---
java/org/apache/catalina/tribes/ByteMessage.java | 4 +-
java/org/apache/catalina/tribes/Channel.java | 56 +++++++---------------
.../apache/catalina/tribes/ChannelException.java | 5 +-
.../apache/catalina/tribes/ChannelInterceptor.java | 2 +-
.../apache/catalina/tribes/ChannelReceiver.java | 8 ++--
.../apache/catalina/tribes/MembershipService.java | 3 +-
.../catalina/tribes/group/ChannelCoordinator.java | 4 +-
.../apache/catalina/tribes/group/RpcChannel.java | 4 +-
.../interceptors/FragmentationInterceptor.java | 2 +-
.../tribes/group/interceptors/GzipInterceptor.java | 23 ++++-----
.../interceptors/MessageDispatchInterceptor.java | 9 +++-
.../group/interceptors/NonBlockingCoordinator.java | 17 ++-----
.../group/interceptors/TcpPingInterceptor.java | 10 +---
.../interceptors/TcpPingInterceptorMBean.java | 7 +++
.../group/interceptors/ThroughputInterceptor.java | 8 +++-
.../interceptors/ThroughputInterceptorMBean.java | 2 +-
.../interceptors/TwoPhaseCommitInterceptor.java | 2 +-
.../org/apache/catalina/tribes/io/ChannelData.java | 19 ++++----
.../org/apache/catalina/tribes/io/XByteBuffer.java | 8 +++-
.../tribes/membership/LocalStrings.properties | 2 +-
.../catalina/tribes/membership/McastService.java | 3 ++
.../tribes/membership/McastServiceImpl.java | 32 ++++---------
.../catalina/tribes/membership/MemberImpl.java | 15 +++++-
.../tribes/membership/MembershipProviderBase.java | 4 +-
.../membership/cloud/CloudMembershipProvider.java | 1 +
.../membership/cloud/DNSMembershipProvider.java | 3 +-
.../cloud/KubernetesMembershipProvider.java | 11 ++---
.../membership/cloud/LocalStrings.properties | 1 +
.../catalina/tribes/transport/ReceiverBase.java | 5 +-
29 files changed, 129 insertions(+), 141 deletions(-)
diff --git a/java/org/apache/catalina/tribes/ByteMessage.java
b/java/org/apache/catalina/tribes/ByteMessage.java
index 4793ef105f..27e4a6d3ad 100644
--- a/java/org/apache/catalina/tribes/ByteMessage.java
+++ b/java/org/apache/catalina/tribes/ByteMessage.java
@@ -50,9 +50,9 @@ public class ByteMessage implements Externalizable {
}
/**
- * Creates a byte message with
+ * Creates a byte message with the specified contents.
*
- * @param data byte[] - the message contents
+ * @param data the message contents
*/
public ByteMessage(byte[] data) {
message = data;
diff --git a/java/org/apache/catalina/tribes/Channel.java
b/java/org/apache/catalina/tribes/Channel.java
index 8c71351059..b7323c64ec 100644
--- a/java/org/apache/catalina/tribes/Channel.java
+++ b/java/org/apache/catalina/tribes/Channel.java
@@ -139,12 +139,7 @@ public interface Channel {
int MBR_TX_SEQ = 8;
/**
- * Send options, when a message is sent, it can have an option flag to
trigger certain behavior. Most flags are used
- * to trigger channel interceptors as the message passes through the
channel stack.
- * <p>
- * However, there are five default flags that every channel implementation
must implement.
- * <p>
- * SEND_OPTIONS_BYTE_MESSAGE - The message is a pure byte message and no
marshaling or unmarshalling will be
+ * Send options. The message is a pure byte message and no marshaling or
unmarshalling will be
* performed.
*
* @see #send(Member[], Serializable , int)
@@ -153,12 +148,7 @@ public interface Channel {
int SEND_OPTIONS_BYTE_MESSAGE = 0x0001;
/**
- * Send options, when a message is sent, it can have an option flag to
trigger certain behavior. Most flags are used
- * to trigger channel interceptors as the message passes through the
channel stack.
- * <p>
- * However, there are five default flags that every channel implementation
must implement
- * <p>
- * SEND_OPTIONS_USE_ACK - Message is sent and an ACK is received when the
message has been received by the
+ * Send options. Message is sent and an ACK is received when the message
has been received by the
* recipient. If no ack is received, the message is not considered
successful.
*
* @see #send(Member[], Serializable , int)
@@ -167,12 +157,7 @@ public interface Channel {
int SEND_OPTIONS_USE_ACK = 0x0002;
/**
- * Send options, when a message is sent, it can have an option flag to
trigger certain behavior. Most flags are used
- * to trigger channel interceptors as the message passes through the
channel stack.
- * <p>
- * However, there are five default flags that every channel implementation
must implement
- * <p>
- * SEND_OPTIONS_SYNCHRONIZED_ACK - Message is sent and an ACK is received
when the message has been received and
+ * Send options. Message is sent and an ACK is received when the message
has been received and
* processed by the recipient. If no ack is received, the message is not
considered successful
*
* @see #send(Member[], Serializable , int)
@@ -181,12 +166,7 @@ public interface Channel {
int SEND_OPTIONS_SYNCHRONIZED_ACK = 0x0004;
/**
- * Send options, when a message is sent, it can have an option flag to
trigger certain behavior. Most flags are used
- * to trigger channel interceptors as the message passes through the
channel stack.
- * <p>
- * However, there are five default flags that every channel implementation
must implement
- * <p>
- * SEND_OPTIONS_ASYNCHRONOUS - Message will be placed on a queue and sent
by a separate thread. If the queue is
+ * Send options. Message will be placed on a queue and sent by a separate
thread. If the queue is
* full, behaviour depends on {@link
MessageDispatchInterceptor#isAlwaysSend()}
*
* @see #send(Member[], Serializable , int)
@@ -195,12 +175,7 @@ public interface Channel {
int SEND_OPTIONS_ASYNCHRONOUS = 0x0008;
/**
- * Send options, when a message is sent, it can have an option flag to
trigger certain behavior. Most flags are used
- * to trigger channel interceptors as the message passes through the
channel stack.
- * <p>
- * However, there are five default flags that every channel implementation
must implement
- * <p>
- * SEND_OPTIONS_SECURE - Message is sent over an encrypted channel
+ * Send options. Message is sent over an encrypted channel
*
* @see #send(Member[], Serializable , int)
* @see #send(Member[], Serializable, int, ErrorHandler)
@@ -225,10 +200,10 @@ public interface Channel {
int SEND_OPTIONS_MULTICAST = 0x0040;
/**
- * Send options, when a message is sent, it can have an option flag to
trigger certain behavior. Most flags are used
+ * When a message is sent, it can have an option flag to trigger certain
behavior. Most flags are used
* to trigger channel interceptors as the message passes through the
channel stack.
* <p>
- * However, there are five default flags that every channel implementation
must implement
+ * However, there are seven default flags that every channel
implementation must implement
* <p>
* SEND_OPTIONS_DEFAULT - the default sending options, just a helper
variable. The default is
* <code>SEND_OPTIONS_USE_ACK</code>
@@ -291,16 +266,16 @@ public interface Channel {
* <li>SND_RX_SEQ - stops the replication receiver</li>
* </ul>
*
- * @throws ChannelException if a startup error occurs or the service is
already stopped or an error occurs.
+ * @throws ChannelException if a shutdown error occurs or the service is
already stopped or an error occurs.
*/
void stop(int svc) throws ChannelException;
/**
* Send a message to one or more members in the cluster
*
- * @param destination Member[] - the destinations, cannot be null or zero
length, the reason for that is that a
- * membership change can occur and at that time the
application is uncertain what group the
- * message actually got sent to.
+ * @param destination Member[] - the destinations, null or zero length
means all members. Note that a membership
+ * change can occur between obtaining the member
list and sending the message, so the
+ * application is uncertain what group the message
actually got sent to.
* @param msg Serializable - the message to send, has to be
serializable, or a <code>ByteMessage</code> to
* send a pure byte array
* @param options int - sender options, see class documentation for
each interceptor that is configured in order
@@ -320,7 +295,9 @@ public interface Channel {
/**
* Send a message to one or more members in the cluster
*
- * @param destination Member[] - the destinations, null or zero length
means all
+ * @param destination Member[] - the destinations, null or zero length
means all members. Note that a membership
+ * change can occur between obtaining the member
list and sending the message, so the
+ * application is uncertain what group the message
actually got sent to.
* @param msg ClusterMessage - the message to send
* @param options int - sender options, see class documentation
* @param handler ErrorHandler - handle errors through a callback,
rather than throw it
@@ -510,6 +487,9 @@ public interface Channel {
* @return a bitwise ORd value of the passed option names
*/
static int parseSendOptions(String input) {
+ if (input == null || input.isEmpty()) {
+ return 0;
+ }
try {
return Integer.parseInt(input);
@@ -537,7 +517,7 @@ public interface Channel {
*
* @param input the int value of SendOptions
*
- * @return the human-friendly string representation in a reverse order
(i.e. the last option will be shown first)
+ * @return the human-friendly comma-separated string representation of the
set options
*/
static String getSendOptionsAsString(int input) {
diff --git a/java/org/apache/catalina/tribes/ChannelException.java
b/java/org/apache/catalina/tribes/ChannelException.java
index 740120ff55..b90424bb19 100644
--- a/java/org/apache/catalina/tribes/ChannelException.java
+++ b/java/org/apache/catalina/tribes/ChannelException.java
@@ -16,6 +16,7 @@
*/
package org.apache.catalina.tribes;
+import java.io.Serializable;
import java.util.ArrayList;
/**
@@ -165,7 +166,9 @@ public class ChannelException extends Exception {
/**
* Represent a failure to a specific member when a message was sent to
more than one member
*/
- public static class FaultyMember {
+ public static class FaultyMember implements Serializable {
+ private static final long serialVersionUID = 1L;
+
/**
* The cause of the failure.
*/
diff --git a/java/org/apache/catalina/tribes/ChannelInterceptor.java
b/java/org/apache/catalina/tribes/ChannelInterceptor.java
index 7c478579cb..ed9acdb23e 100644
--- a/java/org/apache/catalina/tribes/ChannelInterceptor.java
+++ b/java/org/apache/catalina/tribes/ChannelInterceptor.java
@@ -194,7 +194,7 @@ public interface ChannelInterceptor extends
MembershipListener, Heartbeat {
* <li>Channel.SND_RX_SEQ - stops the replication
receiver</li>
* </ul>
*
- * @throws ChannelException if a startup error occurs or the service is
already started.
+ * @throws ChannelException if a shutdown error occurs or the service is
already stopped.
*
* @see Channel
*/
diff --git a/java/org/apache/catalina/tribes/ChannelReceiver.java
b/java/org/apache/catalina/tribes/ChannelReceiver.java
index e14797f390..24105d6e67 100644
--- a/java/org/apache/catalina/tribes/ChannelReceiver.java
+++ b/java/org/apache/catalina/tribes/ChannelReceiver.java
@@ -20,8 +20,8 @@ import java.io.IOException;
/**
* The <code>ChannelReceiver</code> interface is the data receiver component
at the bottom layer, the IO layer (for
- * layers see the {@link Channel} interface). This class may optionally
implement a thread pool for parallel processing
- * of incoming messages.
+ * layers see the {@link Channel} interface). An implementation of this
interface may optionally implement a thread
+ * pool for parallel processing of incoming messages.
*/
public interface ChannelReceiver extends Heartbeat {
/**
@@ -71,11 +71,9 @@ public interface ChannelReceiver extends Heartbeat {
int getUdpPort();
/**
- * Sets the message listener to receive notification of incoming
+ * Sets the message listener to receive notification of incoming messages.
*
* @param listener MessageListener
- *
- * @see MessageListener
*/
void setMessageListener(MessageListener listener);
diff --git a/java/org/apache/catalina/tribes/MembershipService.java
b/java/org/apache/catalina/tribes/MembershipService.java
index 7227188d27..b029ded6ec 100644
--- a/java/org/apache/catalina/tribes/MembershipService.java
+++ b/java/org/apache/catalina/tribes/MembershipService.java
@@ -68,8 +68,7 @@ public interface MembershipService {
/**
- * Stops the membership service. If a membership listeners is added the
listener will start to receive membership
- * events.
+ * Stops the membership service. If a membership listener is added the
listener will stop receiving membership events.
*
* @param level - level MBR_RX stops listening for members, level MBR_TX
stops broadcasting the server
*
diff --git a/java/org/apache/catalina/tribes/group/ChannelCoordinator.java
b/java/org/apache/catalina/tribes/group/ChannelCoordinator.java
index 49ca602a97..f5e950ee09 100644
--- a/java/org/apache/catalina/tribes/group/ChannelCoordinator.java
+++ b/java/org/apache/catalina/tribes/group/ChannelCoordinator.java
@@ -93,7 +93,7 @@ public class ChannelCoordinator extends
ChannelInterceptorBase implements Messag
*
* @param destination Member[] - the destinations, null or zero length
means all
* @param msg ClusterMessage - the message to send
- * @param payload TBA
+ * @param payload InterceptorPayload - optional payload carrying error
handlers and metadata
*/
@Override
public void sendMessage(Member[] destination, ChannelMessage msg,
InterceptorPayload payload)
@@ -219,7 +219,7 @@ public class ChannelCoordinator extends
ChannelInterceptorBase implements Messag
* SND_TX_SEQ - stops the replication transmitter<BR>
* SND_RX_SEQ - stops the replication receiver<BR>
*
- * @throws ChannelException if a startup error occurs or the service is
already started.
+ * @throws ChannelException if a shutdown error occurs or the service is
already stopped.
*/
protected synchronized void internalStop(int svc) throws ChannelException {
try {
diff --git a/java/org/apache/catalina/tribes/group/RpcChannel.java
b/java/org/apache/catalina/tribes/group/RpcChannel.java
index 482ddda1ef..a1ef2bd1a9 100644
--- a/java/org/apache/catalina/tribes/group/RpcChannel.java
+++ b/java/org/apache/catalina/tribes/group/RpcChannel.java
@@ -300,7 +300,7 @@ public class RpcChannel implements ChannelListener {
}
/**
- * Class that holds all response.
+ * Class that holds all responses.
*/
public static class RpcCollector {
/**
@@ -414,7 +414,7 @@ public class RpcChannel implements ChannelListener {
@Override
public int hashCode() {
- return id[0] + id[1] + id[2] + id[3];
+ return java.util.Arrays.hashCode(id);
}
@Override
diff --git
a/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java
b/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java
index 51e0f093ef..a0d9e2eaee 100644
---
a/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java
+++
b/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java
@@ -322,7 +322,7 @@ public class FragmentationInterceptor extends
ChannelInterceptorBase implements
@Override
public int hashCode() {
- return XByteBuffer.toInt(uniqueId, 0);
+ return Arrays.hashCode(uniqueId);
}
@Override
diff --git
a/java/org/apache/catalina/tribes/group/interceptors/GzipInterceptor.java
b/java/org/apache/catalina/tribes/group/interceptors/GzipInterceptor.java
index ab0dc33c9d..0e7ecc7160 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/GzipInterceptor.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/GzipInterceptor.java
@@ -160,10 +160,10 @@ public class GzipInterceptor extends
ChannelInterceptorBase implements GzipInter
*/
public static byte[] compress(byte[] data) throws IOException {
ByteArrayOutputStream bout = new ByteArrayOutputStream();
- GZIPOutputStream gout = new GZIPOutputStream(bout);
- gout.write(data);
- gout.flush();
- gout.close();
+ try (GZIPOutputStream gout = new GZIPOutputStream(bout)) {
+ gout.write(data);
+ gout.finish();
+ }
return bout.toByteArray();
}
@@ -177,13 +177,14 @@ public class GzipInterceptor extends
ChannelInterceptorBase implements GzipInter
*/
public static byte[] decompress(byte[] data) throws IOException {
ByteArrayOutputStream bout = new
ByteArrayOutputStream(DEFAULT_BUFFER_SIZE);
- ByteArrayInputStream bin = new ByteArrayInputStream(data);
- GZIPInputStream gin = new GZIPInputStream(bin);
- byte[] tmp = new byte[DEFAULT_BUFFER_SIZE];
- int length = gin.read(tmp);
- while (length > -1) {
- bout.write(tmp, 0, length);
- length = gin.read(tmp);
+ try (ByteArrayInputStream bin = new ByteArrayInputStream(data);
+ GZIPInputStream gin = new GZIPInputStream(bin)) {
+ byte[] tmp = new byte[DEFAULT_BUFFER_SIZE];
+ int length = gin.read(tmp);
+ while (length > -1) {
+ bout.write(tmp, 0, length);
+ length = gin.read(tmp);
+ }
}
return bout.toByteArray();
}
diff --git
a/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java
b/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java
index b00df219dc..f3af235295 100644
---
a/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java
+++
b/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java
@@ -17,6 +17,7 @@
package org.apache.catalina.tribes.group.interceptors;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@@ -131,8 +132,12 @@ public class MessageDispatchInterceptor extends
ChannelInterceptorBase implement
* @return true if added
*/
public boolean addToQueue(final ChannelMessage msg, final Member[]
destination, final InterceptorPayload payload) {
- executor.execute(() -> sendAsyncData(msg, destination, payload));
- return true;
+ try {
+ executor.execute(() -> sendAsyncData(msg, destination, payload));
+ return true;
+ } catch (RejectedExecutionException ree) {
+ return false;
+ }
}
diff --git
a/java/org/apache/catalina/tribes/group/interceptors/NonBlockingCoordinator.java
b/java/org/apache/catalina/tribes/group/interceptors/NonBlockingCoordinator.java
index 673d9cd030..9c126aa817 100644
---
a/java/org/apache/catalina/tribes/group/interceptors/NonBlockingCoordinator.java
+++
b/java/org/apache/catalina/tribes/group/interceptors/NonBlockingCoordinator.java
@@ -116,18 +116,6 @@ import org.apache.juli.logging.LogFactory;
* <p>
* Ideally, the interceptor below this one would be the TcpFailureDetector to
ensure correct memberships
* </p>
- * <p>
- * The example above, of course can be simplified with a finite
statemachine:<br>
- * But I suck at writing state machines, my head gets all confused. One day I
will document this algorithm though.<br>
- * Maybe I'll do a state diagram :)
- * </p>
- * <h2>State Diagrams</h2>
- * <a
href="https://people.apache.org/~fhanik/tribes/docs/leader-election-initiate-election.jpg";>Initiate
an
- * election</a><br>
- * <br>
- * <a
href="https://people.apache.org/~fhanik/tribes/docs/leader-election-message-arrives.jpg";>Receive
an election
- * message</a><br>
- * <br>
*/
public class NonBlockingCoordinator extends ChannelInterceptorBase {
@@ -730,6 +718,9 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
@Override
public void memberDisappeared(Member member) {
+ if (membership == null) {
+ setupMembership();
+ }
membership.removeMember(member);
super.memberDisappeared(member);
try {
@@ -1012,7 +1003,7 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
* Writes the message to the buffer.
*/
public void write() {
- buf.reset();
+ buf.clear();
// header
buf.append(COORD_HEADER, 0, COORD_HEADER.length);
// leader
diff --git
a/java/org/apache/catalina/tribes/group/interceptors/TcpPingInterceptor.java
b/java/org/apache/catalina/tribes/group/interceptors/TcpPingInterceptor.java
index 1c6bf6a449..28bca893de 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/TcpPingInterceptor.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/TcpPingInterceptor.java
@@ -180,11 +180,7 @@ public class TcpPingInterceptor extends
ChannelInterceptorBase implements TcpPin
return useThread;
}
- /**
- * Returns whether static-only mode is enabled.
- *
- * @return true if only static members are pinged
- */
+ @Override
public boolean getStaticOnly() {
return staticOnly;
}
@@ -263,9 +259,7 @@ public class TcpPingInterceptor extends
ChannelInterceptorBase implements TcpPin
sleep(interval);
sendPing();
} catch (InterruptedException ix) {
- // Ignore. Probably triggered by a call to stop().
- // In the highly unlikely event it was a different trigger,
- // simply ignore it and continue.
+ // Expected during shutdown. The running flag is already
set to false, so the loop will exit.
} catch (Exception e) {
log.warn(sm.getString("tcpPingInterceptor.pingFailed.pingThread"), e);
}
diff --git
a/java/org/apache/catalina/tribes/group/interceptors/TcpPingInterceptorMBean.java
b/java/org/apache/catalina/tribes/group/interceptors/TcpPingInterceptorMBean.java
index ef0f8bf2db..ea77d4a420 100644
---
a/java/org/apache/catalina/tribes/group/interceptors/TcpPingInterceptorMBean.java
+++
b/java/org/apache/catalina/tribes/group/interceptors/TcpPingInterceptorMBean.java
@@ -42,4 +42,11 @@ public interface TcpPingInterceptorMBean {
*/
boolean getUseThread();
+ /**
+ * Returns whether static-only mode is enabled.
+ *
+ * @return true if only static members are pinged
+ */
+ boolean getStaticOnly();
+
}
diff --git
a/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java
b/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java
index 8ca8cac6ce..f063b0d0be 100644
---
a/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java
+++
b/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java
@@ -111,9 +111,13 @@ public class ThroughputInterceptor extends
ChannelInterceptorBase implements Thr
@Override
public void report(double timeTx) {
if (log.isInfoEnabled()) {
+ double txRate = timeTx > 0 ? mbTx / timeTx : 0;
+ double appTxRate = timeTx > 0 ? mbAppTx / timeTx : 0;
+ double rxElapsed = (System.currentTimeMillis() - rxStart) / 1000d;
+ double rxRate = rxElapsed > 0 ? mbRx / rxElapsed : 0;
log.info(sm.getString("throughputInterceptor.report", msgTxCnt,
df.format(mbTx), df.format(mbAppTx),
- df.format(timeTx), df.format(mbTx / timeTx),
df.format(mbAppTx / timeTx), msgTxErr, msgRxCnt,
- df.format(mbRx / ((System.currentTimeMillis() - rxStart) /
(double) 1000)), df.format(mbRx)));
+ df.format(timeTx), df.format(txRate),
df.format(appTxRate), msgTxErr, msgRxCnt,
+ df.format(rxRate), df.format(mbRx)));
}
}
diff --git
a/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptorMBean.java
b/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptorMBean.java
index ee5d4cd2c3..94231e328c 100644
---
a/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptorMBean.java
+++
b/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptorMBean.java
@@ -24,7 +24,7 @@ import java.util.concurrent.atomic.AtomicLong;
public interface ThroughputInterceptorMBean {
/**
- * Returns the socket option flag.
+ * Returns the option flag used to determine if this interceptor should
process a message.
* @return the option flag
*/
int getOptionFlag();
diff --git
a/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java
b/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java
index ce78bd7a7c..2344e0baef 100644
---
a/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java
+++
b/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java
@@ -82,7 +82,7 @@ public class TwoPhaseCommitInterceptor extends
ChannelInterceptorBase {
} else {
confirmation = (ChannelMessage) msg.clone();
}
- confirmation.getMessage().reset();
+ confirmation.getMessage().clear();
UUIDGenerator.randomUUID(false, confirmation.getUniqueId(), 0);
confirmation.getMessage().append(START_DATA, 0, START_DATA.length);
confirmation.getMessage().append(msg.getUniqueId(), 0,
msg.getUniqueId().length);
diff --git a/java/org/apache/catalina/tribes/io/ChannelData.java
b/java/org/apache/catalina/tribes/io/ChannelData.java
index 99a26d5fe4..2f60b376f5 100644
--- a/java/org/apache/catalina/tribes/io/ChannelData.java
+++ b/java/org/apache/catalina/tribes/io/ChannelData.java
@@ -172,9 +172,9 @@ public class ChannelData implements ChannelMessage {
4 + // unique id length off=12
uniqueId.length + // id data off=12+uniqueId.length
4 + // addr length off=12+uniqueId.length+4
- address.getDataLength() + // member data
off=12+uniqueId.length+4+add.length
+ ((address != null) ? address.getDataLength() : 0) + // member
data off=12+uniqueId.length+4+add.length
4 + // message length off=12+uniqueId.length+4+add.length+4
- message.getLength();
+ ((message != null) ? message.getLength() : 0);
}
/**
@@ -198,7 +198,8 @@ public class ChannelData implements ChannelMessage {
* @return the byte array
*/
public byte[] getDataPackage(byte[] data, int offset) {
- byte[] addr = address.getData(false);
+ byte[] addr = (address != null) ? address.getData(false) : new byte[0];
+ int msgLen = (message != null) ? message.getLength() : 0;
XByteBuffer.toBytes(options, data, offset);
offset += 4; // options
XByteBuffer.toBytes(timestamp, data, offset);
@@ -211,9 +212,11 @@ public class ChannelData implements ChannelMessage {
offset += 4; // addr.length
System.arraycopy(addr, 0, data, offset, addr.length);
offset += addr.length; // addr data
- XByteBuffer.toBytes(message.getLength(), data, offset);
+ XByteBuffer.toBytes(msgLen, data, offset);
offset += 4; // message.length
- System.arraycopy(message.getBytesDirect(), 0, data, offset,
message.getLength());
+ if (msgLen > 0) {
+ System.arraycopy(message.getBytesDirect(), 0, data, offset,
msgLen);
+ }
return data;
}
@@ -278,7 +281,6 @@ public class ChannelData implements ChannelMessage {
// data.message = new XByteBuffer(new byte[xsize],false);
data.message = BufferPool.getBufferPool().getBuffer(xsize, false);
offset += 4; // message length
- System.arraycopy(b, offset, data.message.getBytesDirect(), 0, xsize);
data.message.append(b, offset, xsize);
offset += xsize; // message data
return data;
@@ -286,7 +288,7 @@ public class ChannelData implements ChannelMessage {
@Override
public int hashCode() {
- return XByteBuffer.toInt(getUniqueId(), 0);
+ return Arrays.hashCode(getUniqueId());
}
/**
@@ -308,7 +310,7 @@ public class ChannelData implements ChannelMessage {
/**
* Create a shallow clone, only the data gets recreated
*
- * @return ClusterData
+ * @return ChannelData clone
*/
@Override
public ChannelData clone() {
@@ -321,6 +323,7 @@ public class ChannelData implements ChannelMessage {
}
if (this.message != null) {
clone.message = new XByteBuffer(this.message.getBytesDirect(),
false);
+ clone.message.setLength(this.message.getLength());
}
return clone;
}
diff --git a/java/org/apache/catalina/tribes/io/XByteBuffer.java
b/java/org/apache/catalina/tribes/io/XByteBuffer.java
index 28762b9896..8df2165aa6 100644
--- a/java/org/apache/catalina/tribes/io/XByteBuffer.java
+++ b/java/org/apache/catalina/tribes/io/XByteBuffer.java
@@ -154,6 +154,7 @@ public class XByteBuffer implements Serializable {
/**
* Resets the buffer length to zero.
*/
+ @Deprecated
public void reset() {
bufSize = 0;
}
@@ -179,7 +180,7 @@ public class XByteBuffer implements Serializable {
}
/**
- * Resets the buffer
+ * Resets the buffer length to zero.
*/
public void clear() {
bufSize = 0;
@@ -196,6 +197,9 @@ public class XByteBuffer implements Serializable {
* something, or the length of data is 0
*/
public boolean append(ByteBuffer b, int len) {
+ if (len == 0) {
+ return false;
+ }
int newcount = bufSize + len;
if (newcount > buf.length) {
expand(newcount);
@@ -524,7 +528,7 @@ public class XByteBuffer implements Serializable {
/**
* Convert eight bytes to a long
*
- * @param b - the byte array containing the four bytes
+ * @param b - the byte array containing the eight bytes
* @param off - the offset
*
* @return the long value constructed from the eight bytes
diff --git a/java/org/apache/catalina/tribes/membership/LocalStrings.properties
b/java/org/apache/catalina/tribes/membership/LocalStrings.properties
index a4848fb09e..70c20154e9 100644
--- a/java/org/apache/catalina/tribes/membership/LocalStrings.properties
+++ b/java/org/apache/catalina/tribes/membership/LocalStrings.properties
@@ -39,7 +39,6 @@ mcastServiceImpl.memberDisappeared.failed=Unable to process
member disappeared m
mcastServiceImpl.memberExpire=Expire member [{0}]
mcastServiceImpl.memberShutdown=Member [{0}] has shutdown
mcastServiceImpl.messageError=Unable to decode message
-mcastServiceImpl.packet.tooLong=Multicast packet received was too long,
dropping package:[{0}]
mcastServiceImpl.receive.running=McastService.receive already running.
mcastServiceImpl.recovery=Tribes membership running recovery thread.
Multicasting is not functional.
mcastServiceImpl.recovery.failed=Recovery attempt number [{0}] failed. Trying
again in [{1}] seconds.
@@ -56,6 +55,7 @@ mcastServiceImpl.unableReceive.broadcastMessage=Unable to
receive broadcast mess
mcastServiceImpl.waitForMembers.done=Done sleeping, membership established,
start level:[{0}]
mcastServiceImpl.waitForMembers.start=Sleeping for [{0}] milliseconds to
establish cluster membership, start level:[{1}]
+memberImpl.invalid.hostLength=Invalid host length [{0}]
memberImpl.invalid.package.begin=Invalid package, should start with:[{0}]
memberImpl.invalid.package.end=Invalid package, should end with:[{0}]
memberImpl.large.payload=Payload is too large for tribes to handle.
diff --git a/java/org/apache/catalina/tribes/membership/McastService.java
b/java/org/apache/catalina/tribes/membership/McastService.java
index c9a4b06c1c..d873139415 100644
--- a/java/org/apache/catalina/tribes/membership/McastService.java
+++ b/java/org/apache/catalina/tribes/membership/McastService.java
@@ -456,6 +456,7 @@ public class McastService extends MembershipServiceBase
implements MessageListen
@Override
public void broadcast(ChannelMessage message) throws ChannelException {
+ McastServiceImpl impl = this.impl;
if (impl == null || (impl.startLevel & Channel.MBR_TX_SEQ) !=
Channel.MBR_TX_SEQ) {
throw new ChannelException(sm.getString("mcastService.noStart"));
}
@@ -520,6 +521,7 @@ public class McastService extends MembershipServiceBase
implements MessageListen
if (localMember != null) {
localMember.setPayload(payload);
try {
+ McastServiceImpl impl = this.impl;
if (impl != null) {
impl.send(false);
}
@@ -535,6 +537,7 @@ public class McastService extends MembershipServiceBase
implements MessageListen
if (localMember != null) {
localMember.setDomain(domain);
try {
+ McastServiceImpl impl = this.impl;
if (impl != null) {
impl.send(false);
}
diff --git a/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
b/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
index b6fcff0d5e..0a4a854a90 100644
--- a/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
+++ b/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
@@ -42,8 +42,7 @@ import org.apache.juli.logging.LogFactory;
* A <b>membership</b> implementation using simple multicast. This is the
representation of a multicast membership
* service. This class is responsible for maintaining a list of active cluster
nodes in the cluster. If a node fails to
* send out a heartbeat, the node will be dismissed. This is the low level
implementation that handles the multicasting
- * sockets. Need to fix this, could use java.nio and only need one thread to
send and receive, or just use a timeout on
- * the receive
+ * sockets.
*/
public class McastServiceImpl extends MembershipProviderBase {
@@ -94,10 +93,6 @@ public class McastServiceImpl extends MembershipProviderBase
{
* How often to we send out a broadcast saying we are alive, must be
smaller than timeToExpiration
*/
protected final long sendFrequency;
- /**
- * Reuse the sendPacket, no need to create a new one every time
- */
- protected DatagramPacket sendPacket;
/**
* Reuse the receivePacket, no need to create a new one every time
*/
@@ -196,9 +191,6 @@ public class McastServiceImpl extends
MembershipProviderBase {
*/
public void init() throws IOException {
setupSocket();
- sendPacket = new DatagramPacket(new byte[MAX_PACKET_SIZE],
MAX_PACKET_SIZE);
- sendPacket.setAddress(address);
- sendPacket.setPort(port);
receivePacket = new DatagramPacket(new byte[MAX_PACKET_SIZE],
MAX_PACKET_SIZE);
receivePacket.setAddress(address);
receivePacket.setPort(port);
@@ -374,23 +366,19 @@ public class McastServiceImpl extends
MembershipProviderBase {
}
/**
- * Receive a datagram packet, locking wait
+ * Receive a datagram packet from the multicast socket with timeout.
*
* @throws IOException Received failed
*/
public void receive() throws IOException {
try {
socket.receive(receivePacket);
- if (receivePacket.getLength() > MAX_PACKET_SIZE) {
- log.error(sm.getString("mcastServiceImpl.packet.tooLong",
Integer.toString(receivePacket.getLength())));
+ byte[] data = new byte[receivePacket.getLength()];
+ System.arraycopy(receivePacket.getData(),
receivePacket.getOffset(), data, 0, data.length);
+ if (XByteBuffer.firstIndexOf(data, 0, MemberImpl.TRIBES_MBR_BEGIN)
== 0) {
+ memberDataReceived(data);
} else {
- byte[] data = new byte[receivePacket.getLength()];
- System.arraycopy(receivePacket.getData(),
receivePacket.getOffset(), data, 0, data.length);
- if (XByteBuffer.firstIndexOf(data, 0,
MemberImpl.TRIBES_MBR_BEGIN) == 0) {
- memberDataReceived(data);
- } else {
- memberBroadcastsReceived(data);
- }
+ memberBroadcastsReceived(data);
}
} catch (SocketTimeoutException ignore) {
/*
@@ -555,10 +543,10 @@ public class McastServiceImpl extends
MembershipProviderBase {
} else if (log.isTraceEnabled()) {
log.trace("Sending message broadcast " + packet.getLength() + "
bytes from " + member);
}
- packet.setAddress(address);
- packet.setPort(port);
// TODO this operation is not thread safe
synchronized (sendLock) {
+ packet.setAddress(address);
+ packet.setPort(port);
socket.send(packet);
}
if (checkexpired) {
@@ -732,7 +720,7 @@ public class McastServiceImpl extends
MembershipProviderBase {
protected static class RecoveryThread extends Thread {
/**
- * Flag to track if recovery is running.
+ * Flag to track if recovery is running. Only one recovery can run at
a time.
*/
private static final AtomicBoolean running = new AtomicBoolean(false);
diff --git a/java/org/apache/catalina/tribes/membership/MemberImpl.java
b/java/org/apache/catalina/tribes/membership/MemberImpl.java
index e93ce04e67..cc556fecfe 100644
--- a/java/org/apache/catalina/tribes/membership/MemberImpl.java
+++ b/java/org/apache/catalina/tribes/membership/MemberImpl.java
@@ -184,7 +184,7 @@ public class MemberImpl implements Member,
java.io.Externalizable {
/**
* Create a data package to send over the wire representing this member.
This is faster than serialization.
*
- * @return - the bytes for this member deserialized
+ * @return the serialized bytes for this member
*/
public byte[] getData() {
return getData(true);
@@ -281,6 +281,10 @@ public class MemberImpl implements Member,
java.io.Externalizable {
XByteBuffer.toBytes(udpPort, data, pos);
pos += 4;
// host length
+ if (host.length > 127) {
+ throw new
IllegalArgumentException(sm.getString("memberImpl.invalid.hostLength",
+ Integer.toString(host.length)));
+ }
data[pos++] = (byte) host.length;
// host
System.arraycopy(host, 0, data, pos, host.length);
@@ -399,6 +403,10 @@ public class MemberImpl implements Member,
java.io.Externalizable {
byte hl = data[pos++];
+ if (hl < 0) {
+ throw new
IllegalArgumentException(sm.getString("memberImpl.invalid.hostLength",
+ Integer.toString(hl)));
+ }
byte[] addr = new byte[hl];
System.arraycopy(data, pos, addr, 0, hl);
pos += hl;
@@ -611,7 +619,10 @@ public class MemberImpl implements Member,
java.io.Externalizable {
@Override
public int hashCode() {
- return getHost()[0] + getHost()[1] + getHost()[2] + getHost()[3];
+ int result = Arrays.hashCode(getHost());
+ result = 31 * result + getPort();
+ result = 31 * result + Arrays.hashCode(getUniqueId());
+ return result;
}
/**
diff --git
a/java/org/apache/catalina/tribes/membership/MembershipProviderBase.java
b/java/org/apache/catalina/tribes/membership/MembershipProviderBase.java
index ec185f4544..0cfa5026af 100644
--- a/java/org/apache/catalina/tribes/membership/MembershipProviderBase.java
+++ b/java/org/apache/catalina/tribes/membership/MembershipProviderBase.java
@@ -67,7 +67,7 @@ public abstract class MembershipProviderBase implements
MembershipProvider {
@Override
public Member getMember(Member mbr) {
- if (membership.getMembers() == null) {
+ if (membership == null || membership.getMembers() == null) {
return null;
}
return membership.getMember(mbr);
@@ -75,7 +75,7 @@ public abstract class MembershipProviderBase implements
MembershipProvider {
@Override
public Member[] getMembers() {
- if (membership.getMembers() == null) {
+ if (membership == null || membership.getMembers() == null) {
return Membership.EMPTY_MEMBERS;
}
return membership.getMembers();
diff --git
a/java/org/apache/catalina/tribes/membership/cloud/CloudMembershipProvider.java
b/java/org/apache/catalina/tribes/membership/cloud/CloudMembershipProvider.java
index 0c130aec6a..ee61f4f8c9 100644
---
a/java/org/apache/catalina/tribes/membership/cloud/CloudMembershipProvider.java
+++
b/java/org/apache/catalina/tribes/membership/cloud/CloudMembershipProvider.java
@@ -188,6 +188,7 @@ public abstract class CloudMembershipProvider extends
MembershipProviderBase imp
@Override
public boolean stop(int level) throws Exception {
+ service.getChannel().removeChannelListener(this);
return true;
}
diff --git
a/java/org/apache/catalina/tribes/membership/cloud/DNSMembershipProvider.java
b/java/org/apache/catalina/tribes/membership/cloud/DNSMembershipProvider.java
index d9be130e9f..e43d66b3b2 100644
---
a/java/org/apache/catalina/tribes/membership/cloud/DNSMembershipProvider.java
+++
b/java/org/apache/catalina/tribes/membership/cloud/DNSMembershipProvider.java
@@ -185,7 +185,7 @@ public class DNSMembershipProvider extends
CloudMembershipProvider {
try {
member = new MemberImpl(ip, port, aliveTime);
} catch (IOException ioe) {
-
log.error(sm.getString("kubernetesMembershipProvider.memberError"), ioe);
+
log.error(sm.getString("dnsMembershipProvider.memberError"), ioe);
continue;
}
member.setUniqueId(id);
@@ -227,6 +227,7 @@ public class DNSMembershipProvider extends
CloudMembershipProvider {
member.setMemberAliveTime(-1);
updateMember(member, true);
}
+ // We will not actually process the message
return false;
}
}
diff --git
a/java/org/apache/catalina/tribes/membership/cloud/KubernetesMembershipProvider.java
b/java/org/apache/catalina/tribes/membership/cloud/KubernetesMembershipProvider.java
index 83d6206ca8..410b103fbb 100644
---
a/java/org/apache/catalina/tribes/membership/cloud/KubernetesMembershipProvider.java
+++
b/java/org/apache/catalina/tribes/membership/cloud/KubernetesMembershipProvider.java
@@ -76,6 +76,9 @@ public class KubernetesMembershipProvider extends
CloudMembershipProvider {
}
String protocol = getEnv(CUSTOM_ENV_PREFIX + "MASTER_PROTOCOL",
"KUBERNETES_MASTER_PROTOCOL");
+ if (protocol == null) {
+ protocol = "https";
+ }
String masterHost = getEnv(CUSTOM_ENV_PREFIX + "MASTER_HOST",
"KUBERNETES_SERVICE_HOST");
String masterPort = getEnv(CUSTOM_ENV_PREFIX + "MASTER_PORT",
"KUBERNETES_SERVICE_PORT");
@@ -87,9 +90,6 @@ public class KubernetesMembershipProvider extends
CloudMembershipProvider {
}
if (clientCertificateFile == null) {
- if (protocol == null) {
- protocol = "https";
- }
String saTokenFile = getEnv(CUSTOM_ENV_PREFIX + "SA_TOKEN_FILE",
"SA_TOKEN_FILE");
if (saTokenFile == null) {
saTokenFile =
"/var/run/secrets/kubernetes.io/serviceaccount/token";
@@ -103,9 +103,6 @@ public class KubernetesMembershipProvider extends
CloudMembershipProvider {
log.error(sm.getString("kubernetesMembershipProvider.streamError"), ioe);
}
} else {
- if (protocol == null) {
- protocol = "http";
- }
String clientKeyFile = getEnv("KUBERNETES_CLIENT_KEY_FILE");
if (clientKeyFile == null) {
log.error(sm.getString("kubernetesMembershipProvider.noKey"));
@@ -243,7 +240,7 @@ public class KubernetesMembershipProvider extends
CloudMembershipProvider {
Object objectUid = metadata.get("uid");
Object creationTimestampObject =
metadata.get("creationTimestamp");
if (creationTimestampObject == null) {
-
log.warn(sm.getString("kubernetesMembershipProvider.invalidPod", "uid"));
+
log.warn(sm.getString("kubernetesMembershipProvider.invalidPod",
"creationTimestamp"));
continue;
}
// "status" contains "phase" (which must be "Running") and
"podIP"
diff --git
a/java/org/apache/catalina/tribes/membership/cloud/LocalStrings.properties
b/java/org/apache/catalina/tribes/membership/cloud/LocalStrings.properties
index 632f49ccc6..0a0816cf02 100644
--- a/java/org/apache/catalina/tribes/membership/cloud/LocalStrings.properties
+++ b/java/org/apache/catalina/tribes/membership/cloud/LocalStrings.properties
@@ -32,6 +32,7 @@ cloudMembershipProvider.start=Namespace [{0}] set; clustering
enabled
cloudMembershipService.stopFail=Unable to stop the cloud membership service,
level: [{0}]
dnsMembershipProvider.dnsError=Error getting hosts address list for namespace
[{0}]
+dnsMembershipProvider.memberError=Error creating member
kubernetesMembershipProvider.invalidPod=Pod is missing some required
attributes: [{0}]
kubernetesMembershipProvider.invalidPodsList=Invalid pods list: [{0}]
diff --git a/java/org/apache/catalina/tribes/transport/ReceiverBase.java
b/java/org/apache/catalina/tribes/transport/ReceiverBase.java
index 3f45ff7016..7f52950caf 100644
--- a/java/org/apache/catalina/tribes/transport/ReceiverBase.java
+++ b/java/org/apache/catalina/tribes/transport/ReceiverBase.java
@@ -139,10 +139,7 @@ public abstract class ReceiverBase implements
ChannelReceiver, ListenCallback, R
}
executor = null;
if (oname != null) {
- JmxRegistry jmxRegistry = JmxRegistry.getRegistry(channel);
- if (jmxRegistry != null) {
- jmxRegistry.unregisterJmx(oname);
- }
+ JmxRegistry.getRegistry(channel).unregisterJmx(oname);
oname = null;
}
channel = null;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
![https://people.apache.org/~fhanik/tribes/docs/leader-election-initiate-election.jpg"](https://people.apache.org/~fhanik/tribes/docs/leader-election-initiate-election.jpg"){kind=link}
![https://people.apache.org/~fhanik/tribes/docs/leader-election-message-arrives.jpg"](https://people.apache.org/~fhanik/tribes/docs/leader-election-message-arrives.jpg"){kind=link}