[feat] [broker] PIP-188: Support option to disconnect clients that not support cluster migration feature

conf/broker.conf

@@ -302,6 +302,12 @@ isAllowAutoUpdateSchemaEnabled=true
 # Enable check for minimum allowed client library version
 clientLibraryVersionCheckEnabled=false
 
+# Minimum client version allowed by broker else broker will reject connection.
+# (It's useful when client lib doesn't support specific feature and feature 
+# might be required by broker to apply globally on all topics.
+# (eg: all clients must be on V20 to perform cloud migration)
+clientMinVersionAllowed=-1
+
 # Path for the file used to determine the rotation status for the broker when responding
 # to service discovery health checks
 statusFilePath=

conf/standalone.conf

@@ -191,6 +191,12 @@ isAllowAutoUpdateSchemaEnabled=true
 # Enable check for minimum allowed client library version
 clientLibraryVersionCheckEnabled=false
 
+# Minimum client version allowed by broker else broker will reject connection.
+# (It's useful when client lib doesn't support specific feature and feature 
+# might be required by broker to apply globally on all topics.
+# (eg: all clients must be on V20 to perform cloud migration)
+clientMinVersionAllowed=-1
+
 # Path for the file used to determine the rotation status for the broker when responding
 # to service discovery health checks
 statusFilePath=/usr/local/apache/htdocs

deployment/terraform-ansible/templates/broker.conf

@@ -205,6 +205,12 @@ maxTopicsPerNamespace=0
 # Enable check for minimum allowed client library version
 clientLibraryVersionCheckEnabled=false
 
+# Minimum client version allowed by broker else broker will reject connection.
+# (It's useful when client lib doesn't support specific feature and feature 
+# might be required by broker to apply globally on all topics.
+# (eg: all clients must be on V20 to perform cloud migration)
+clientMinVersionAllowed=-1
+
 # Path for the file used to determine the rotation status for the broker when responding
 # to service discovery health checks
 statusFilePath=

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -850,6 +850,17 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
         doc = "Enable check for minimum allowed client library version"
     )
     private boolean clientLibraryVersionCheckEnabled = false;
+
+    @FieldContext(
+        category = CATEGORY_SERVER,
+        dynamic = true,
+        doc = "Minimum client version allowed by broker else broker will reject connection."
+              + "(It's useful when client lib doesn't support specific feature and feature "
+              + "might be required by broker to apply globally on all topics."
+              + "(eg: all clients must be on V20 to perform cloud migration)"
+    )
+    private int clientMinVersionAllowed = -1;
+
     @FieldContext(
         category = CATEGORY_SERVER,
         doc = "Path for the file used to determine the rotation status for the broker"

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -189,6 +189,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
     private final TopicListService topicListService;
     private final BrokerInterceptor brokerInterceptor;
     private State state;
+    private int clientMinVersionAllowed;
     private volatile boolean isActive = true;
     private String authRole = null;
     private volatile AuthenticationDataSource authenticationData;
@@ -307,6 +308,7 @@ public ServerCnx(PulsarService pulsar, String listenerName) {
         this.topicListService = new TopicListService(pulsar, this,
                 enableSubscriptionPatternEvaluation, maxSubscriptionPatternLength);
         this.brokerInterceptor = this.service != null ? this.service.getInterceptor() : null;
+        this.clientMinVersionAllowed = conf.getClientMinVersionAllowed() > 0 ? conf.getClientMinVersionAllowed() : -1;
     }
 
     @Override
@@ -684,6 +686,14 @@ ByteBuf createConsumerStatsResponse(Consumer consumer, long requestId) {
 
     // complete the connect and sent newConnected command
     private void completeConnect(int clientProtoVersion, String clientVersion) {
+        if (clientMinVersionAllowed >= 0 && clientMinVersionAllowed > clientProtoVersion) {
+            log.info("[{}] client with version {} must be upgraded to {}", remoteAddress, clientProtoVersion,
+                    clientMinVersionAllowed);
+            final ByteBuf msg = Commands.newError(-1, ServerError.UnsupportedVersionError,
+                    "Upgrade version to " + clientMinVersionAllowed + " or higher");
+            NettyChannelUtil.writeAndFlushWithClosePromise(ctx, msg);
+            return;
+        }
         if (service.isAuthenticationEnabled()) {
             if (service.isAuthorizationEnabled()) {
                 if (!service.getAuthorizationService()

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java

@@ -21,17 +21,18 @@
 
 import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs;
 import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgsRecordingInvocations;
+import static org.junit.Assert.assertNotEquals;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.matches;
 import static org.mockito.Mockito.CALLS_REAL_METHODS;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.matches;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -42,11 +43,7 @@
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
-import io.netty.channel.ChannelHandler;
-import io.netty.channel.embedded.EmbeddedChannel;
-import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
+
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.nio.charset.StandardCharsets;
@@ -64,6 +61,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
+
 import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.CloseCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCursorCallback;
@@ -76,14 +74,13 @@
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.TransactionMetadataStoreService;
 import org.apache.pulsar.broker.auth.MockAlwaysExpiredAuthenticationProvider;
+import org.apache.pulsar.broker.auth.MockAuthenticationProvider;
 import org.apache.pulsar.broker.auth.MockAuthorizationProvider;
+import org.apache.pulsar.broker.auth.MockMultiStageAuthenticationProvider;
 import org.apache.pulsar.broker.auth.MockMutableAuthenticationProvider;
 import org.apache.pulsar.broker.authentication.AuthenticationDataSubscription;
-import org.apache.pulsar.broker.testcontext.PulsarTestContext;
-import org.apache.pulsar.broker.TransactionMetadataStoreService;
-import org.apache.pulsar.broker.auth.MockAuthenticationProvider;
-import org.apache.pulsar.broker.auth.MockMultiStageAuthenticationProvider;
 import org.apache.pulsar.broker.authentication.AuthenticationProvider;
 import org.apache.pulsar.broker.authentication.AuthenticationService;
 import org.apache.pulsar.broker.authorization.AuthorizationService;
@@ -93,6 +90,7 @@
 import org.apache.pulsar.broker.service.ServerCnx.State;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.broker.service.utils.ClientChannelHelper;
+import org.apache.pulsar.broker.testcontext.PulsarTestContext;
 import org.apache.pulsar.client.api.transaction.TxnID;
 import org.apache.pulsar.common.api.AuthData;
 import org.apache.pulsar.common.api.proto.AuthMethod;
@@ -149,6 +147,12 @@
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
+
 @SuppressWarnings("unchecked")
 @Test(groups = "broker")
 public class ServerCnxTest {
@@ -251,6 +255,27 @@ public void testConnectCommand() throws Exception {
         channel.finish();
     }
 
+    @Test(timeOut = 30000)
+    public void testClientMinVersionSupport() throws Exception {
+        int clientMinVersion = 10;
+        int clientVersion = 9;
+        pulsar.getConfiguration().setClientMinVersionAllowed(clientMinVersion);
+        resetChannel();
+        setConnectionVersion(clientVersion);
+        assertTrue(channel.isActive());
+        assertEquals(serverCnx.getState(), State.Start);
+
+        // test server response to CONNECT
+        ByteBuf clientCommand = Commands.newConnect("none", null, clientVersion, "", "", "", null, null, null);
+        channel.writeInbound(clientCommand);
+
+        assertNotEquals(serverCnx.getState(), State.Connected);
+        Object error = getResponse();
+        assertTrue(error instanceof CommandError);
+        assertTrue(((CommandError) error).getError().equals(ServerError.UnsupportedVersionError));
+        channel.finish();
+    }
+
     private static ByteBuf newConnect(AuthMethod authMethod, String authData, int protocolVersion) {
         BaseCommand cmd = new BaseCommand().setType(Type.CONNECT);
         cmd.setConnect()