You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by al...@apache.org on 2024/01/19 22:42:58 UTC
(kudu) branch branch-1.17.x updated (587dbc0af -> df022eeb1)
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a change to branch branch-1.17.x
in repository https://gitbox.apache.org/repos/asf/kudu.git
from 587dbc0af KUDU-3532: Follow up to replica placement bug
new 8af89ca5e KUDU-3491 Destruct master before creating a new one
new df022eeb1 KUDU-3526 [java] Scanner should bind with a tserver in java client.
The 2 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:
.../org/apache/kudu/client/AsyncKuduClient.java | 11 +--
.../org/apache/kudu/client/AsyncKuduScanner.java | 12 +++
.../java/org/apache/kudu/client/RemoteTablet.java | 11 +++
.../org/apache/kudu/client/TestKuduScanner.java | 89 ++++++++++++++++++++++
src/kudu/master/dynamic_multi_master-test.cc | 44 ++++++++++-
src/kudu/master/master_runner.cc | 1 +
6 files changed, 159 insertions(+), 9 deletions(-)
(kudu) 01/02: KUDU-3491 Destruct master before creating a new one
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch branch-1.17.x
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 8af89ca5e2454600888873882ce543cdb0b7b3ab
Author: Ádám Bakai <ab...@cloudera.com>
AuthorDate: Wed Jan 17 15:43:59 2024 +0100
KUDU-3491 Destruct master before creating a new one
ServerBase constructor runs MinidumpExceptionHandler constructor that
calls RegisterMinidumpExceptionHandler(). This function increments the
static atomic variable current_num_instances_. Then the ServerBase is
destructed, a similar process happens and current_num_instances_ gets
decremented. If current_num_instances_ is not zero before incrementing
or not 1 before decrementing, then it is considered an error. This
indicates that only one Server can run at any given time. But in case of
multi-master config, the master server is replaced, and without the
change it is possible that the second server's constructor precede first
server's destructor. This change makes it sure that the destructor is
executed before the second one's constructor.
Change-Id: I3c1019d092bbf9e58f4fc33753a1218bc79735d3
Reviewed-on: http://gerrit.cloudera.org:8080/20913
Reviewed-by: Attila Bukor <ab...@apache.org>
Reviewed-by: Mahesh Reddy <mr...@cloudera.com>
Tested-by: Kudu Jenkins
(cherry picked from commit 7562277fc6f68b0dcab593d56de03bb344a95b3e)
Reviewed-on: http://gerrit.cloudera.org:8080/20917
Tested-by: Alexey Serbin <al...@apache.org>
Reviewed-by: Abhishek Chennaka <ac...@cloudera.com>
---
src/kudu/master/dynamic_multi_master-test.cc | 44 ++++++++++++++++++++++++++--
src/kudu/master/master_runner.cc | 1 +
2 files changed, 43 insertions(+), 2 deletions(-)
diff --git a/src/kudu/master/dynamic_multi_master-test.cc b/src/kudu/master/dynamic_multi_master-test.cc
index d605fe0d3..28bd92b78 100644
--- a/src/kudu/master/dynamic_multi_master-test.cc
+++ b/src/kudu/master/dynamic_multi_master-test.cc
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+#include <kudu/gutil/strings/util.h>
+
#include <algorithm>
#include <atomic>
#include <cstdint>
@@ -28,6 +30,7 @@
#include <string>
#include <thread>
#include <tuple>
+#include <type_traits>
#include <unordered_set>
#include <utility>
#include <vector>
@@ -35,7 +38,6 @@
#include <gflags/gflags.h>
#include <glog/logging.h>
#include <gtest/gtest.h>
-#include <kudu/gutil/strings/util.h>
#include "kudu/client/client.h"
#include "kudu/client/schema.h"
@@ -1456,7 +1458,7 @@ struct MultiMasterClusterArgs {
class AutoAddMasterTest : public KuduTest {
public:
- Status SetUpWithTestArgs(const MultiMasterClusterArgs& args) {
+ virtual Status SetUpWithTestArgs(const MultiMasterClusterArgs& args) {
opts_.num_masters = args.orig_num_masters;
opts_.enable_kerberos = args.is_secure;
args_ = args;
@@ -1477,6 +1479,17 @@ class AutoAddMasterTest : public KuduTest {
unique_ptr<ExternalMiniCluster> cluster_;
};
+class MinidumpTest : public AutoAddMasterTest {
+ public:
+ Status SetUpWithTestArgs(const MultiMasterClusterArgs& args) override {
+ // By default minidump is disabled for all ExternalMiniCluster instances.
+ // This made a bug undetected during testing, so a positive regression test
+ // was added where minidump is enabled.
+ opts_.extra_master_flags.emplace_back("--enable_minidumps=true");
+ return AutoAddMasterTest::SetUpWithTestArgs(args);
+ }
+};
+
constexpr const int64_t kShortRetryIntervalSecs = 1;
// Test that nothing goes wrong when starting up masters but the entire cluster
@@ -1777,6 +1790,33 @@ TEST_F(AutoAddMasterTest, TestAddNewMaster) {
NO_FATALS(cluster_->AssertNoCrashes());
}
+// Regression test for KUDU-3491
+TEST_F(MinidumpTest, TestAddNewMasterMinidumpsEnabled) {
+ SKIP_IF_SLOW_NOT_ALLOWED();
+
+ // Let's get the current master addresses and add a new one to them
+ vector<HostPort> master_addrs = cluster_->master_rpc_addrs();
+ unique_ptr<Socket> reserved_socket;
+ ASSERT_OK(MiniCluster::ReserveDaemonSocket(
+ MiniCluster::DaemonType::MASTER, master_addrs.size(), opts_.bind_mode, &reserved_socket));
+
+ Sockaddr addr;
+ ASSERT_OK(reserved_socket->GetSocketAddress(&addr));
+ master_addrs.emplace_back(addr.host(), addr.port());
+
+ // Let's create the new master and start it to ensure it starts up okay.
+ scoped_refptr<ExternalMaster> peer;
+ auto idx = cluster_->master_rpc_addrs().size();
+ ASSERT_OK(cluster_->CreateMaster(master_addrs, idx, &peer));
+ ASSERT_OK(peer->Start());
+ ASSERT_OK(peer->WaitForCatalogManager());
+ auto expected_num_masters = ++idx;
+ ASSERT_EVENTUALLY([&] {
+ ASSERT_OK(VerifyVotersOnAllMasters(expected_num_masters, cluster_.get()));
+ });
+ NO_FATALS(cluster_->AssertNoCrashes());
+}
+
class ParameterizedAutoAddMasterTest : public AutoAddMasterTest,
public ::testing::WithParamInterface<tuple<int, bool>> {
public:
diff --git a/src/kudu/master/master_runner.cc b/src/kudu/master/master_runner.cc
index 60a7cc68b..4861ffc60 100644
--- a/src/kudu/master/master_runner.cc
+++ b/src/kudu/master/master_runner.cc
@@ -455,6 +455,7 @@ Status RunMasterServer() {
// If we succeeded, wipe the system catalog on this node and initiate a
// copy from another node.
RETURN_NOT_OK(ClearLocalSystemCatalogAndCopy(leader_hp));
+ server.reset();
server.reset(new Master(opts));
RETURN_NOT_OK(server->Init());
RETURN_NOT_OK(server->Start());
(kudu) 02/02: KUDU-3526 [java] Scanner should bind with a tserver in java client.
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch branch-1.17.x
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit df022eeb1da6717310b72ab63466e6f0c629ebd2
Author: 宋家成 <so...@thinkingdata.cn>
AuthorDate: Mon Nov 20 16:49:22 2023 +0800
KUDU-3526 [java] Scanner should bind with a tserver in java client.
For now, scan requests sent by the java client might fail with scanner
not found error. It is because that scanner does not bind with the
tserver with which it first communicates.
The code in method scanNextRows of java client is still trying to
search for the tserver by the locations and selection policy. So if
the leader is changed and the next scan request is sent to the new
leader, the tserver will respond with the "scanner not found"
exception.
We should bind the scanner to the tserver, similar to how it is done
in the C++ client.
Change-Id: I9cf65f4215e99198dd41b43d14e50c8c23b8a9b2
Reviewed-on: http://gerrit.cloudera.org:8080/20715
Tested-by: Kudu Jenkins
Reviewed-by: Yifan Zhang <ch...@163.com>
Reviewed-by: Alexey Serbin <al...@apache.org>
(cherry picked from commit 3895f4f796579071615addfa12d2abf75753b9a1)
Reviewed-on: http://gerrit.cloudera.org:8080/20923
Tested-by: Alexey Serbin <al...@apache.org>
Reviewed-by: Abhishek Chennaka <ac...@cloudera.com>
---
.../org/apache/kudu/client/AsyncKuduClient.java | 11 +--
.../org/apache/kudu/client/AsyncKuduScanner.java | 12 +++
.../java/org/apache/kudu/client/RemoteTablet.java | 11 +++
.../org/apache/kudu/client/TestKuduScanner.java | 89 ++++++++++++++++++++++
4 files changed, 116 insertions(+), 7 deletions(-)
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
index b9ada6de4..84ffedf97 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
@@ -1325,8 +1325,7 @@ public class AsyncKuduClient implements AutoCloseable {
// Important to increment the attempts before the next if statement since
// getSleepTimeForRpc() relies on it if the client is null or dead.
nextRequest.attempt++;
- final ServerInfo info = tablet.getReplicaSelectedServerInfo(nextRequest.getReplicaSelection(),
- location);
+ final ServerInfo info = tablet.getTabletServerByUuid(scanner.getTsUUID());
if (info == null) {
return delayedSendRpcToTablet(nextRequest, new RecoverableException(Status.RemoteError(
String.format("No information on servers hosting tablet %s, will retry later",
@@ -1352,8 +1351,7 @@ public class AsyncKuduClient implements AutoCloseable {
return Deferred.fromResult(null);
}
final KuduRpc<AsyncKuduScanner.Response> closeRequest = scanner.getCloseRequest();
- final ServerInfo info = tablet.getReplicaSelectedServerInfo(closeRequest.getReplicaSelection(),
- location);
+ final ServerInfo info = tablet.getTabletServerByUuid(scanner.getTsUUID());
if (info == null) {
return Deferred.fromResult(null);
}
@@ -1381,8 +1379,7 @@ public class AsyncKuduClient implements AutoCloseable {
}
final KuduRpc<Void> keepAliveRequest = scanner.getKeepAliveRequest();
- final ServerInfo info =
- tablet.getReplicaSelectedServerInfo(keepAliveRequest.getReplicaSelection(), location);
+ final ServerInfo info = tablet.getTabletServerByUuid(scanner.getTsUUID());
if (info == null) {
return Deferred.fromResult(null);
}
@@ -1436,7 +1433,7 @@ public class AsyncKuduClient implements AutoCloseable {
if (entry != null) {
RemoteTablet tablet = entry.getTablet();
ServerInfo info = tablet.getReplicaSelectedServerInfo(request.getReplicaSelection(),
- location);
+ getLocationString());
if (info != null) {
Deferred<R> d = request.getDeferred();
request.setTablet(tablet);
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
index 4929dd4d8..779b368dd 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
@@ -284,6 +284,12 @@ public final class AsyncKuduScanner {
final long scanRequestTimeout;
+ /**
+ * UUID of the tserver which the scanner is bound with. The following scans of
+ * this scanner will be sent to the tserver.
+ */
+ private String tsUUID;
+
/**
* The prefetching result is cached in memory. This atomic reference is used to avoid
* two concurrent prefetchings occur and the latest one overrides the previous one.
@@ -541,6 +547,10 @@ public final class AsyncKuduScanner {
this.rowDataFormat = rowDataFormat;
}
+ public String getTsUUID() {
+ return tsUUID;
+ }
+
/**
* Scans a number of rows.
* <p>
@@ -599,12 +609,14 @@ public final class AsyncKuduScanner {
}
if (!resp.more || resp.scannerId == null) {
+ tsUUID = resp.data.getTsUUID();
scanFinished();
return Deferred.fromResult(resp.data); // there might be data to return
}
scannerId = resp.scannerId;
sequenceId++;
canRequestMore = resp.more;
+ tsUUID = resp.data.getTsUUID();
if (LOG.isDebugEnabled()) {
LOG.debug("Scanner {} opened on {}", Bytes.pretty(scannerId), tablet);
}
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java b/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java
index 753093729..2c035df74 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java
@@ -302,6 +302,17 @@ public class RemoteTablet implements Comparable<RemoteTablet> {
return results;
}
+ /**
+ * Get information on tablet server by its UUID.
+ * @param uuid tablet server uuid.
+ * @return tablet server info by the uuid.
+ */
+ ServerInfo getTabletServerByUuid(String uuid) {
+ synchronized (tabletServers) {
+ return tabletServers.get(uuid);
+ }
+ }
+
@Override
public int compareTo(RemoteTablet remoteTablet) {
if (remoteTablet == null) {
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduScanner.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduScanner.java
index 54e7d6577..4225ad6b0 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduScanner.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduScanner.java
@@ -17,10 +17,13 @@
package org.apache.kudu.client;
+import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.kudu.client.AsyncKuduScanner.DEFAULT_IS_DELETED_COL_NAME;
+import static org.apache.kudu.test.ClientTestUtil.createManyStringsSchema;
import static org.apache.kudu.test.ClientTestUtil.getBasicCreateTableOptions;
import static org.apache.kudu.test.ClientTestUtil.getBasicSchema;
import static org.apache.kudu.test.ClientTestUtil.loadDefaultTable;
+import static org.apache.kudu.test.junit.AssertHelpers.assertEventuallyTrue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
@@ -38,6 +41,7 @@ import java.util.Map;
import java.util.Random;
import java.util.Set;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import org.junit.Before;
import org.junit.Rule;
@@ -54,6 +58,7 @@ import org.apache.kudu.test.CapturingLogAppender;
import org.apache.kudu.test.KuduTestHarness;
import org.apache.kudu.test.RandomUtils;
import org.apache.kudu.test.cluster.KuduBinaryLocator;
+import org.apache.kudu.test.junit.AssertHelpers;
import org.apache.kudu.util.DataGenerator;
import org.apache.kudu.util.Pair;
@@ -561,4 +566,88 @@ public class TestKuduScanner {
assertTrue(row.hasIsDeleted());
assertTrue(row.isDeleted());
}
+
+ @Test
+ public void testScannerLeaderChanged() throws Exception {
+ // Prepare the table for testing.
+ Schema schema = createManyStringsSchema();
+ CreateTableOptions createOptions = new CreateTableOptions();
+ final int buckets = 2;
+ createOptions.addHashPartitions(ImmutableList.of("key"), buckets);
+ createOptions.setNumReplicas(3);
+ client.createTable(tableName, schema, createOptions);
+
+ KuduSession session = client.newSession();
+ KuduTable table = client.openTable(tableName);
+ final int totalRows = 2000;
+ for (int i = 0; i < totalRows; i++) {
+ Insert insert = table.newInsert();
+ PartialRow row = insert.getRow();
+ row.addString("key", String.format("key_%02d", i));
+ row.addString("c1", "c1_" + i);
+ row.addString("c2", "c2_" + i);
+ assertEquals(session.apply(insert).hasRowError(), false);
+ }
+ AsyncKuduClient asyncClient = harness.getAsyncClient();
+ KuduScanner kuduScanner = new KuduScanner.KuduScannerBuilder(asyncClient, table)
+ .replicaSelection(ReplicaSelection.LEADER_ONLY)
+ .batchSizeBytes(100)
+ .build();
+
+ // Open the scanner first.
+ kuduScanner.nextRows();
+ final HostAndPort referenceServerHostPort = harness.findLeaderTabletServer(
+ new LocatedTablet(kuduScanner.currentTablet()));
+ final String referenceTabletId = kuduScanner.currentTablet().getTabletId();
+
+ // Send LeaderStepDown request.
+ KuduBinaryLocator.ExecutableInfo exeInfo = KuduBinaryLocator.findBinary("kudu");
+ LOG.info(harness.getMasterAddressesAsString());
+ List<String> commandLine = Lists.newArrayList(exeInfo.exePath(),
+ "tablet",
+ "leader_step_down",
+ harness.findLeaderMasterServer().toString(),
+ kuduScanner.currentTablet().getTabletId());
+ ProcessBuilder processBuilder = new ProcessBuilder(commandLine);
+ processBuilder.environment().putAll(exeInfo.environment());
+ Process stepDownProcess = processBuilder.start();
+ assertEquals(0, stepDownProcess.waitFor());
+
+ // Wait until the leader changes.
+ assertEventuallyTrue(
+ "The leadership should be transferred",
+ new AssertHelpers.BooleanExpression() {
+ @Override
+ public boolean get() throws Exception {
+ asyncClient.emptyTabletsCacheForTable(table.getTableId());
+ List<LocatedTablet> tablets = table.getTabletsLocations(50000);
+ LocatedTablet targetTablet = null;
+ for (LocatedTablet tablet : tablets) {
+ String tabletId = new String(tablet.getTabletId(), UTF_8);
+ if (tabletId.equals(referenceTabletId)) {
+ targetTablet = tablet;
+ }
+ }
+ HostAndPort targetHp = harness.findLeaderTabletServer(targetTablet);
+ return !targetHp.equals(referenceServerHostPort);
+ }
+ },
+ 10000/*timeoutMillis*/);
+
+ // Simulate that another request(like Batch) has sent to the wrong leader tablet server and
+ // the change of leadership has been acknowledged. The response will demote the leader.
+ kuduScanner.currentTablet().demoteLeader(
+ kuduScanner.currentTablet().getLeaderServerInfo().getUuid());
+ asyncClient.emptyTabletsCacheForTable(table.getTableId());
+
+ int rowsScannedInNextScans = 0;
+ try {
+ while (kuduScanner.hasMoreRows()) {
+ rowsScannedInNextScans += kuduScanner.nextRows().numRows;
+ }
+ } catch (Exception ex) {
+ assertFalse(ex.getMessage().matches(".*Scanner .* not found.*"));
+ }
+ assertTrue(rowsScannedInNextScans > 0);
+ }
}