Skip to content

Commit 9d398da

Browse files
yyxjcccchen6lishanglin
authored
Feature/hetero support non gtid replication (#701)
* rdb parse distinct non-gtid and gtid not parsed * xsync replication support offset * fix xsync reconnect loss init gtid_excluded * applier support subenv * skip merge start & end when rdb gtidset is empty * move repl conf listening port to xsync replication * modify fws keeper memeory usage * modify fws keeper memeory usage * modify fws keeper memeory usage * fix reset parser bug * destory txnProvider when applier shutdown * optimize code * optmize applier offset accuracy * fix offset accuracy * remove cluster load when destory * remove route when destory * xsync partial sync support offset * fix ut * fix ut * use place holder gtidset in xsync when not support gtid * fix code * merge master * move publish filter to applier * filter publish command adapt non gtid * update gtid state before filter redis op * fix handler leak in rdb only psync * fix handle leak in keeper add/remove * use new credis * destory locator * fix deadlock on keeper slave close * make session loactor impl lifecycle * fix threads not released in keeper 1. Indexing not released 2. slave psync thread hang on offset await * forget running commands and obstacle instead of success * reset threshold park when destory * fix applier ut down * fix keeper integration test --------- Co-authored-by: cchen6 <[email protected]> Co-authored-by: lishanglin <[email protected]>
1 parent 6eeeaf0 commit 9d398da

File tree

81 files changed

+837
-238
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

81 files changed

+837
-238
lines changed

‎core/src/main/java/com/ctrip/xpipe/client/redis/AsyncRedisClientFactory.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -14,9 +14,9 @@ public interface AsyncRedisClientFactory extends Ordered {
1414

1515
AsyncRedisClientFactory DEFAULT = ServicesUtil.getAsyncRedisClientFactory();
1616

17-
AsyncRedisClient getOrCreateClient(String clusterName, ExecutorService credisNotifyExecutor) throws Exception;
17+
AsyncRedisClient getOrCreateClient(String clusterName, String subenv, ExecutorService credisNotifyExecutor) throws Exception;
1818

19-
AsyncRedisClient createClient(String clusterName, ExecutorService credisNotifyExecutor) throws Exception;
19+
AsyncRedisClient createClient(String clusterName, String subenv, ExecutorService credisNotifyExecutor) throws Exception;
2020

2121
default int getOrder() {
2222
return LOWEST_PRECEDENCE;

‎core/src/main/java/com/ctrip/xpipe/client/redis/DoNothingRedisClientFactory.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -10,12 +10,12 @@
1010
public class DoNothingRedisClientFactory implements AsyncRedisClientFactory {
1111

1212
@Override
13-
public AsyncRedisClient getOrCreateClient(String clusterName, ExecutorService credisNotifyExecutor) {
13+
public AsyncRedisClient getOrCreateClient(String clusterName, String subenv, ExecutorService credisNotifyExecutor) {
1414
return new DoNothingRedisClient();
1515
}
1616

1717
@Override
18-
public AsyncRedisClient createClient(String clusterName, ExecutorService credisNotifyExecutor) {
18+
public AsyncRedisClient createClient(String clusterName, String subenv, ExecutorService credisNotifyExecutor) {
1919
return new DoNothingRedisClient();
2020
}
2121
}

‎core/src/main/java/com/ctrip/xpipe/gtid/GtidSet.java

+4
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,10 @@
1212

1313
public class GtidSet {
1414

15+
public static final String EMPTY_GTIDSET = "";
16+
17+
public static final String PLACE_HOLDER = "0:0";
18+
1519
public static Pair<String, Long> parseGtid(String gtid) {
1620
String[] split = gtid.split(":");
1721
if (split.length != 2) {

‎core/src/main/java/com/ctrip/xpipe/lifecycle/CreatedComponentRedistry.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -57,8 +57,6 @@ public boolean doRemove(Object component) throws Exception {
5757
}
5858

5959
logger.info("[doRemove]{}, {}" , name, component);
60-
components.remove(name);
61-
6260
if(component instanceof Lifecycle){
6361
Lifecycle lifecycle = (Lifecycle) component;
6462

@@ -72,6 +70,8 @@ public boolean doRemove(Object component) throws Exception {
7270
}
7371
}
7472
}
73+
74+
components.remove(name);
7575
return true;
7676
}
7777

‎dump.rdb

109 Bytes
Binary file not shown.

‎pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,7 @@
5757
<influxdb.version>2.7</influxdb.version>
5858
<hickwall.client.version>1.10-SNAPSHOT</hickwall.client.version>
5959
<clogging.version>4.6.3</clogging.version>
60-
<credis.version>4.4.1-unstable-4</credis.version>
60+
<credis.version>4.4.1-unstable-SNAPSHOT</credis.version>
6161
<qconfig.sso.version>1.0.7-SNAPSHOT</qconfig.sso.version>
6262
<qunaer.common.version>9.2.37</qunaer.common.version>
6363
<qconfig.client.version>1.100.46-ctrip</qconfig.client.version>

‎redis/package/redis-keeper-package/src/main/scripts/startup.sh

+4-4
Original file line numberDiff line numberDiff line change
@@ -136,10 +136,10 @@ then
136136
JAVA_OPTS="$JAVA_OPTS -Xms${USED_MEM}g -Xmx${USED_MEM}g -Xmn${XMN}g -XX:+AlwaysPreTouch -XX:MaxDirectMemorySize=${MAX_DIRECT}g"
137137
elif [ $ENV = "FWS" ] || [ $ENV = "FAT" ];then
138138
#MB
139-
USED_MEM=600
140-
XMN=450
141-
MAX_DIRECT=100
142-
JAVA_OPTS="$JAVA_OPTS -Xms${USED_MEM}m -Xmx${USED_MEM}m -Xmn${XMN}m -XX:+AlwaysPreTouch -XX:MaxDirectMemorySize=${MAX_DIRECT}m"
139+
USED_MEM=`getSafeXmx`
140+
XMN=`getSafeXmn $USED_MEM`
141+
MAX_DIRECT=2
142+
JAVA_OPTS="$JAVA_OPTS -Xms${USED_MEM}g -Xmx${USED_MEM}g -Xmn${XMN}g -XX:+AlwaysPreTouch -XX:MaxDirectMemorySize=${MAX_DIRECT}g"
143143
else
144144
changeConfigLogFile $FULL_DIR log4j2-uat.xml
145145

‎redis/redis-core/src/main/java/com/ctrip/xpipe/redis/core/entity/ApplierTransMeta.java

+10
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@ public class ApplierTransMeta {
2525

2626
private Long concurrencyThreshold;
2727

28+
private String subenv;
29+
2830
//for json conversion
2931
public ApplierTransMeta() {}
3032

@@ -103,6 +105,14 @@ public void setConcurrencyThreshold(Long concurrencyThreshold) {
103105
this.concurrencyThreshold = concurrencyThreshold;
104106
}
105107

108+
public String getSubenv() {
109+
return subenv;
110+
}
111+
112+
public void setSubenv(String subenv) {
113+
this.subenv = subenv;
114+
}
115+
106116
@Override
107117
public boolean equals(Object o) {
108118
if (this == o) return true;

‎redis/redis-core/src/main/java/com/ctrip/xpipe/redis/core/protocal/PsyncObserver.java

+2
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,8 @@ public interface PsyncObserver {
3030

3131
void readRdbGtidSet(RdbStore rdbStore, String gtidSet);
3232

33+
void readAuxEnd(RdbStore rdbStore);
34+
3335
void endWriteRdb();
3436

3537
void onContinue(String requestReplId, String responseReplId);

‎redis/redis-core/src/main/java/com/ctrip/xpipe/redis/core/protocal/XsyncObserver.java

+5-5
Original file line numberDiff line numberDiff line change
@@ -10,16 +10,16 @@
1010
*/
1111
public interface XsyncObserver {
1212

13-
void onFullSync(GtidSet rdbGtidSet);
13+
void onFullSync(GtidSet rdbGtidSet, long rdbOffset);
1414

15-
void beginReadRdb(EofType eofType, GtidSet rdbGtidSet);
15+
void beginReadRdb(EofType eofType, GtidSet rdbGtidSet, long rdbOffset);
1616

1717
void onRdbData(ByteBuf rdbData);
1818

19-
void endReadRdb(EofType eofType, GtidSet rdbGtidSet);
19+
void endReadRdb(EofType eofType, GtidSet rdbGtidSet, long rdbOffset);
2020

21-
void onContinue(GtidSet gtidSetExcluded);
21+
void onContinue(GtidSet gtidSetExcluded, long continueOffset);
2222

23-
void onCommand(Object[] rawCmdArgs);
23+
void onCommand(long commandOffset, Object[] rawCmdArgs);
2424

2525
}

‎redis/redis-core/src/main/java/com/ctrip/xpipe/redis/core/protocal/cmd/AbstractReplicationStorePsync.java

+13
Original file line numberDiff line numberDiff line change
@@ -135,6 +135,19 @@ public void onAux(String key, String value) {
135135
}
136136
}
137137

138+
@Override
139+
public void onAuxFinish() {
140+
getLogger().info("[onAuxFinish] aux is finish");
141+
142+
for (PsyncObserver observer : observers) {
143+
try {
144+
observer.readAuxEnd(rdbStore);
145+
} catch (Throwable th) {
146+
getLogger().error("[onAuxFinish]" + this, th);
147+
}
148+
}
149+
}
150+
138151
protected void readRdbGtidSet(String gtidSet) {
139152

140153
getLogger().info("[readRdbGtidSet]{}, gtidset:{}", this, gtidSet);

‎redis/redis-core/src/main/java/com/ctrip/xpipe/redis/core/protocal/cmd/DefaultXsync.java

+55-25
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package com.ctrip.xpipe.redis.core.protocal.cmd;
22

3+
import com.ctrip.xpipe.api.command.Command;
34
import com.ctrip.xpipe.api.command.CommandFuture;
45
import com.ctrip.xpipe.api.command.CommandFutureListener;
56
import com.ctrip.xpipe.api.payload.InOutPayload;
@@ -8,10 +9,11 @@
89
import com.ctrip.xpipe.gtid.GtidSet;
910
import com.ctrip.xpipe.lifecycle.LifecycleHelper;
1011
import com.ctrip.xpipe.netty.commands.NettyClient;
11-
import com.ctrip.xpipe.pool.FixedObjectPool;
1212
import com.ctrip.xpipe.pool.ReturnObjectException;
1313
import com.ctrip.xpipe.redis.core.exception.RedisRuntimeException;
14-
import com.ctrip.xpipe.redis.core.protocal.*;
14+
import com.ctrip.xpipe.redis.core.protocal.RedisClientProtocol;
15+
import com.ctrip.xpipe.redis.core.protocal.Xsync;
16+
import com.ctrip.xpipe.redis.core.protocal.XsyncObserver;
1517
import com.ctrip.xpipe.redis.core.protocal.protocal.AbstractBulkStringParser;
1618
import com.ctrip.xpipe.redis.core.protocal.protocal.EofType;
1719
import com.ctrip.xpipe.redis.core.protocal.protocal.RdbBulkStringParser;
@@ -20,14 +22,13 @@
2022
import com.ctrip.xpipe.utils.StringUtil;
2123
import io.netty.buffer.ByteBuf;
2224
import io.netty.channel.Channel;
23-
import io.netty.channel.ChannelFuture;
24-
import io.netty.channel.ChannelFutureListener;
2525

2626
import java.io.IOException;
2727
import java.nio.channels.WritableByteChannel;
2828
import java.util.LinkedList;
2929
import java.util.List;
3030
import java.util.concurrent.ScheduledExecutorService;
31+
import java.util.concurrent.atomic.AtomicLong;
3132

3233
import static com.ctrip.xpipe.redis.core.protocal.Xsync.XSYNC_STATE.READING_COMMANDS;
3334

@@ -47,22 +48,34 @@ public class DefaultXsync extends AbstractRedisCommand<Object> implements Xsync,
4748

4849
private GtidSet rdbDataGtidSet;
4950

51+
private long rdbOffset;
52+
5053
private NettyClient nettyClient;
5154

5255
private List<XsyncObserver> observers = new LinkedList<>();
5356

5457
protected Xsync.XSYNC_STATE xsyncState = XSYNC_STATE.XSYNC_COMMAND_WAIT_RESPONSE;
5558

59+
private int listeningPort;
60+
61+
private AtomicLong currentCommandOffset;
62+
5663
public DefaultXsync(String host, int port, GtidSet gtidSetExcluded, Object vectorClockExcluded, ScheduledExecutorService scheduled) {
5764
super(host, port, scheduled);
5865
this.gitdSetExcluded = gtidSetExcluded;
5966
this.vectorClockExcluded = vectorClockExcluded;
6067
}
6168

62-
public DefaultXsync(SimpleObjectPool<NettyClient> clientPool, GtidSet gtidSetExcluded, Object vectorClockExcluded, ScheduledExecutorService scheduled) {
69+
public DefaultXsync(SimpleObjectPool<NettyClient> clientPool, GtidSet gtidSetExcluded, Object vectorClockExcluded, ScheduledExecutorService scheduled, int listeningPort) {
6370
super(clientPool, scheduled);
6471
this.gitdSetExcluded = gtidSetExcluded;
6572
this.vectorClockExcluded = vectorClockExcluded;
73+
this.listeningPort = listeningPort;
74+
this.currentCommandOffset = new AtomicLong(0);
75+
}
76+
77+
public NettyClient getNettyClient() {
78+
return nettyClient;
6679
}
6780

6881
@Override
@@ -118,9 +131,11 @@ protected Object doReceiveResponse(Channel channel, ByteBuf byteBuf) throws Exce
118131
break;
119132

120133
case READING_COMMANDS:
134+
int prevIndex = byteBuf.readerIndex();
121135
Object cmdPayload = super.doReceiveResponse(channel, byteBuf);
136+
currentCommandOffset.addAndGet(byteBuf.readerIndex() - prevIndex);
122137
if (cmdPayload instanceof Object[]) {
123-
doOnCommand((Object[]) cmdPayload);
138+
doOnCommand(currentCommandOffset.getAndSet(0), (Object[]) cmdPayload);
124139
} else if (null != cmdPayload) {
125140
getLogger().info("[doReceiveResponse][{}][unknown payload] {}, {}", READING_COMMANDS, this, cmdPayload);
126141
throw new RedisRuntimeException("unknown payload:" + cmdPayload);
@@ -153,17 +168,24 @@ protected void handleRedisResponse(Channel channel, String xsync) throws IOExcep
153168
}
154169

155170
if (split[0].equalsIgnoreCase(FULL_SYNC)) {
156-
if (split.length != 2) {
171+
if (split.length < 2) {
157172
throw new RedisRuntimeException("unknown reply:" + xsync);
158173
}
159-
this.rdbDataGtidSet = new GtidSet(split[1]);
174+
this.rdbDataGtidSet = GtidSet.PLACE_HOLDER.equals(split[1]) ? new GtidSet(GtidSet.EMPTY_GTIDSET) : new GtidSet(split[1]);
175+
if (split.length > 2) {
176+
this.rdbOffset = Long.parseLong(split[2]);
177+
}
160178
getLogger().debug("[readRedisResponse][FULL]{}, {} {}", ChannelUtil.getDesc(channel), this, rdbDataGtidSet);
161179
xsyncState = XSYNC_STATE.READING_RDB;
162180
doOnFullSync();
163181
} else if (split[0].equalsIgnoreCase(PARTIAL_SYNC)) {
164182
xsyncState = READING_COMMANDS;
165183
getLogger().debug("[readRedisResponse][PARTIAL]{}, {}", ChannelUtil.getDesc(channel), this);
166-
doOnContinue();
184+
long continueOffset = 0;
185+
if (split.length > 1) {
186+
continueOffset = Long.parseLong(split[1]);
187+
}
188+
doOnContinue(continueOffset);
167189
} else {
168190
throw new RedisRuntimeException("unknown reply:" + xsync);
169191
}
@@ -173,31 +195,31 @@ private void doOnFullSync() {
173195
getLogger().debug("[doOnFullSync] {}", this);
174196
for (XsyncObserver observer: observers) {
175197
try {
176-
observer.onFullSync(rdbDataGtidSet);
198+
observer.onFullSync(rdbDataGtidSet, rdbOffset);
177199
} catch (Throwable th) {
178200
getLogger().error("[doOnFullSync][fail] {}", observer, th);
179201
}
180202
}
181203
resetClient();
182204
}
183205

184-
private void doOnContinue() {
206+
private void doOnContinue(long continueOffset) {
185207
getLogger().debug("[doOnContinue] {}", this);
186208
for (XsyncObserver observer: observers) {
187209
try {
188-
observer.onContinue(gitdSetExcluded);
210+
observer.onContinue(gitdSetExcluded, continueOffset);
189211
} catch (Throwable th) {
190212
getLogger().error("[doOnContinue][fail] {}", observer, th);
191213
}
192214
}
193215
resetClient();
194216
}
195217

196-
private void doOnCommand(Object[] rawCmdArgs) {
218+
private void doOnCommand(long commandOffset, Object[] rawCmdArgs) {
197219
getLogger().debug("[doOnCommand] {}", this);
198220
for (XsyncObserver observer: observers) {
199221
try {
200-
observer.onCommand(rawCmdArgs);
222+
observer.onCommand(commandOffset, rawCmdArgs);
201223
} catch (Throwable th) {
202224
getLogger().error("[doOnCommand][fail] {}", observer, th);
203225
}
@@ -209,7 +231,7 @@ private void beginReadRdb() {
209231
getLogger().debug("[beginReadRdb] {}", this);
210232
for (XsyncObserver observer: observers) {
211233
try {
212-
observer.beginReadRdb(eofType, rdbDataGtidSet);
234+
observer.beginReadRdb(eofType, rdbDataGtidSet, rdbOffset);
213235
} catch (Throwable th) {
214236
getLogger().error("[beginReadRdb][fail] {}", observer, th);
215237
}
@@ -223,36 +245,44 @@ private void onRdbData(ByteBuf byteBuf) {
223245
observer.onRdbData(byteBuf.slice());
224246
} catch (Throwable th) {
225247
getLogger().error("[notifyRdbData][fail] {}", observer, th);
248+
throw th;
226249
}
227250
}
228251
}
229252

230253
private void endReadRdb() {
231254
getLogger().debug("[endReadRdb] {}", this);
232-
if (eofType.putOnLineOnAck()) {
233-
Replconf replconf = new Replconf(new FixedObjectPool<>(nettyClient), Replconf.ReplConfType.ACK, scheduled, "1");
234-
replconf.execute().addListener(commandFuture -> {
235-
if (!commandFuture.isSuccess()) {
236-
getLogger().info("[endReadRdb][ack] fail", commandFuture.cause());
237-
future().setFailure(new XpipeRuntimeException("ack after rdb fail", commandFuture.cause()));
238-
}
239-
});
240-
}
241255

242256
for (XsyncObserver observer: observers) {
243257
try {
244-
observer.endReadRdb(eofType, rdbDataGtidSet);
258+
observer.endReadRdb(eofType, rdbDataGtidSet, rdbOffset);
245259
} catch (Throwable th) {
246260
getLogger().error("[notifyRdbData][fail] {}", observer, th);
247261
}
248262
}
249263
}
250264

265+
private Command<Object> replConfListeningPort() {
266+
267+
return new Replconf(getClientPool(), Replconf.ReplConfType.LISTENING_PORT, scheduled,
268+
String.valueOf(listeningPort));
269+
}
270+
251271
@Override
252272
protected void afterCommandExecute(NettyClient nettyClient) {
253273
// temporary solution, handle channel evicted by channel pool
254274

255275
this.nettyClient = nettyClient;
276+
277+
replConfListeningPort().execute().addListener(new CommandFutureListener<Object>() {
278+
@Override
279+
public void operationComplete(CommandFuture<Object> commandFuture) throws Exception {
280+
if (!commandFuture.isSuccess()) {
281+
close();
282+
}
283+
}
284+
});
285+
256286
nettyClient.channel().closeFuture().addListener(closeFuture -> {
257287
if (!future().isDone()) {
258288
future().setFailure(new XpipeRuntimeException("channel closed"));

‎redis/redis-core/src/main/java/com/ctrip/xpipe/redis/core/redis/rdb/RdbParseListener.java

+2
Original file line numberDiff line numberDiff line change
@@ -14,4 +14,6 @@ public interface RdbParseListener {
1414

1515
void onFinish(RdbParser<?> parser);
1616

17+
void onAuxFinish();
18+
1719
}

0 commit comments

Comments
 (0)