Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

adds shard reconnect mechanic if connection to manager is lost #3425

Merged
merged 4 commits into from
May 7, 2024
Merged

adds shard reconnect mechanic if connection to manager is lost #3425

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
96d480b
adds shard reconnect mechanic if connection to manager is lost
thoniTUB May 7, 2024
483a83e
Merge branch 'develop' into feature/shard-reconnect
thoniTUB May 7, 2024
a3f44fe
fix getValidator NPE
thoniTUB May 7, 2024
db23aac
Merge branch 'develop' into feature/shard-reconnect
thoniTUB May 7, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 46 additions & 10 deletions backend/src/main/java/com/bakdata/conquery/commands/ShardNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import com.bakdata.conquery.models.messages.network.specific.AddShardNode;
import com.bakdata.conquery.models.messages.network.specific.RegisterWorker;
import com.bakdata.conquery.models.messages.network.specific.UpdateJobManagerStatus;
import com.bakdata.conquery.models.worker.IdResolveContext;
import com.bakdata.conquery.models.worker.Worker;
import com.bakdata.conquery.models.worker.WorkerInformation;
import com.bakdata.conquery.models.worker.Workers;
Expand All @@ -51,6 +52,7 @@
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.FilterEvent;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.jetbrains.annotations.NotNull;

/**
* This node holds a shard of data (in so called {@link Worker}s for the different datasets in conquery.
Expand All @@ -64,6 +66,7 @@ public class ShardNode extends ConqueryCommand implements IoHandler, Managed {
public static final String DEFAULT_NAME = "shard-node";

private NioSocketConnector connector;
private ConnectFuture future;
private JobManager jobManager;
private Validator validator;
private ConqueryConfig config;
Expand All @@ -88,7 +91,6 @@ protected void run(Environment environment, Namespace namespace, ConqueryConfig
this.environment = environment;
this.config = config;

connector = new NioSocketConnector();

jobManager = new JobManager(getName(), config.isFailOnError());
environment.lifecycle().manage(this);
Expand All @@ -106,6 +108,7 @@ protected void run(Environment environment, Namespace namespace, ConqueryConfig
getConfig().getQueries().getSecondaryIdSubPlanRetention()
);


final Collection<WorkerStorage> workerStorages = config.getStorage().discoverWorkerStorages();


Expand Down Expand Up @@ -268,6 +271,8 @@ private static void scheduleIdleLogger(ScheduledExecutorService scheduler, IoSes
public void sessionClosed(IoSession session) {
setLocation(session);
log.info("Disconnected from ManagerNode.");

scheduler.schedule(this::connectToCluster, 2, TimeUnit.SECONDS);
}

@Override
Expand All @@ -291,7 +296,9 @@ public void messageSent(IoSession session, Object message) {
@Override
public void inputClosed(IoSession session) {
setLocation(session);
log.info("Session closed.");
log.info("Input closed.");
session.closeNow();
scheduler.schedule(this::disconnectFromCluster, 0, TimeUnit.SECONDS);
}

@Override
Expand All @@ -306,24 +313,27 @@ public void start() throws Exception {
value.getJobManager().addSlowJob(new SimpleJob("Update Bucket Manager", value.getBucketManager()::fullUpdate));
}

ObjectMapper om = createInternalObjectMapper(View.InternalCommunication.class);
scheduler.schedule(this::connectToCluster, 0, TimeUnit.MINUTES);

BinaryJacksonCoder coder = new BinaryJacksonCoder(workers, validator, om);
connector.getFilterChain().addLast("codec", new CQProtocolCodecFilter(new ChunkWriter(coder), new ChunkReader(coder, om)));
connector.setHandler(this);
connector.getSessionConfig().setAll(config.getCluster().getMina());

}

private void connectToCluster() {
InetSocketAddress address = new InetSocketAddress(
config.getCluster().getManagerURL().getHostAddress(),
config.getCluster().getPort()
);

disconnectFromCluster();

connector = getClusterConnector(workers);

while (true) {
try {
log.info("Trying to connect to {}", address);

// Try opening a connection (Note: This fails immediately instead of waiting a minute to try and connect)
ConnectFuture future = connector.connect(address);
future = connector.connect(address);

future.awaitUninterruptibly();

Expand All @@ -339,9 +349,24 @@ public void start() throws Exception {
catch (RuntimeIoException e) {
log.warn("Failed to connect to {}", address, e);
}
catch (InterruptedException e) {
log.warn("Interrupted while trying to connector to cluster, giving up.", e);
break;
}
}
}

@NotNull
private NioSocketConnector getClusterConnector(IdResolveContext workers) {
ObjectMapper om = createInternalObjectMapper(View.InternalCommunication.class);

NioSocketConnector connector = new NioSocketConnector();

BinaryJacksonCoder coder = new BinaryJacksonCoder(workers, validator, om);
connector.getFilterChain().addLast("codec", new CQProtocolCodecFilter(new ChunkWriter(coder), new ChunkReader(coder, om)));
connector.setHandler(this);
connector.getSessionConfig().setAll(config.getCluster().getMina());
return connector;
}

@Override
Expand All @@ -350,12 +375,23 @@ public void stop() throws Exception {

workers.stop();

disconnectFromCluster();
}

private void disconnectFromCluster() {
if (future != null) {
future.cancel();
}

//after the close command was send
if (context != null) {
context.awaitClose();
}
log.info("Connection was closed by ManagerNode");
connector.dispose();

if (connector != null) {
log.info("Connection was closed by ManagerNode");
connector.dispose();
}
}

public boolean isBusy() {
Expand Down
Loading