We are no longer monitoring this channel, please join Slack! https://join.slack.com/t/atomixio/shared_invite/enQtNDgzNjA5MjMyMDUxLTVmMThjZDcxZDE3ZmU4ZGYwZTc2MGJiYjVjMjFkOWMyNmVjYTc5YjExYTZiOWFjODlkYmE2MjNjYzZhNjU2MjY
Atomix 3.0.0-rc6 has been released!
- Stuff
- Things
It’s mostly bug fixes. The only API changes were in
AtomicDocumentTree where we changed the paths from root|this|stupid|format to /this/better/format
It looks like we cannot see rc6 in maven repository imediately
Does
String not in the namespace of ClusterCommunicationService?
It's strange that I have to register a
String Class
It will take a few minutes
You don’t have to register
String
Or many core Java primitives and collections
I got this error
java.util.concurrent.ExecutionException: com.esotericsoftware.kryo.KryoException: Encountered unregistered class ID: 112
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
at cn.ac.iie.di.atomix3test.atomix.n.Client$T.send(Client.java:150)
at cn.ac.iie.di.atomix3test.atomix.n.Client$T.run(Client.java:122)
Caused by: com.esotericsoftware.kryo.KryoException: Encountered unregistered class ID: 112
at com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:137)
at com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:693)
at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:804)
at io.atomix.utils.serializer.Namespace.lambda$null$2(Namespace.java:452)
at com.esotericsoftware.kryo.pool.KryoPoolQueueImpl.run(KryoPoolQueueImpl.java:58)
at io.atomix.utils.serializer.Namespace.lambda$deserialize$3(Namespace.java:450)
at io.atomix.utils.serializer.KryoIOPool.run(KryoIOPool.java:45)
at io.atomix.utils.serializer.Namespace.deserialize(Namespace.java:448)
at io.atomix.utils.serializer.Serializer$1.decode(Serializer.java:57)
at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962)
at io.atomix.cluster.messaging.impl.NettyMessagingService.lambda$null$11(NettyMessagingService.java:491)
at com.google.common.util.concurrent.MoreExecutors$DirectExecutor.execute(MoreExecutors.java:399)
at io.atomix.cluster.messaging.impl.NettyMessagingService.lambda$null$14(NettyMessagingService.java:491)
at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760)
at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736)
at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962)
at io.atomix.cluster.messaging.impl.NettyMessagingService$Callback.complete(NettyMessagingService.java:823)
at io.atomix.cluster.messaging.impl.NettyMessagingService$RemoteClientConnection.dispatch(NettyMessagingService.java:1072)
at io.atomix.cluster.messaging.impl.NettyMessagingService$RemoteClientConnection.access$1000(NettyMessagingService.java:1023)
at io.atomix.cluster.messaging.impl.NettyMessagingService$InboundMessageDispatcher.channelRead0(NettyMessagingService.java:757)
at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:284)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1434)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:965)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsa
and my client is like this:
MemberId leader = election.getLeadership().leader().id();
CompletableFuture<byte[]> future = atomix.getCommunicationService().send("test_type", "this is a msg to leader".getBytes(), leader);
while (!future.isDone()) {
try {
sleep(500);
} catch (InterruptedException ex) {
}
}
try {
System.out.println(this.getName() + " - " + count + " - " + new String(future.get()));
} catch (InterruptedException | ExecutionException ex) {
System.err.println(this.getName() + " - " + count + " - error ");
ex.printStackTrace();
}
while my server do got msg:
coming in test_type, addr:127.0.0.1:16176, cmd: this is a msg to leader
coming in test_type, addr:127.0.0.1:16177, cmd: this is a msg to leader
there is another question, does
ClusterCommunicationService change my msg? I can see more bytes received.
server code:
atomix.getCommunicationService().subscribe("test_type", (byte[] src) -> {
System.out.println("coming in test_type, addr:" + "unknown" + ", cmd:" + new String(src));
try {
Thread.sleep(1000);
} catch (InterruptedException ex) {
}
return ("return_" + i.getAndIncrement() + "_" + new String(src) + "_from_" + server).getBytes();
}, new ThreadPoolExecutor(2, 32, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(32), (r, e) -> r.run()));
you’re sending a byte array, not a string
one sec
I sent a String , failed, and change it to byte[]
and find it failed too
Collection<Node> bootstrapLocations = Arrays.asList(
Node.builder().withId("foo").withAddress(Address.from("localhost:5000")).build(),
Node.builder().withId("bar").withAddress(Address.from("localhost:5001")).build(),
Node.builder().withId("baz").withAddress(Address.from("localhost:5002")).build());
AtomixCluster cluster1 = AtomixCluster.builder()
.withMemberId("foo")
.withAddress("localhost:5000")
.withMembershipProvider(BootstrapDiscoveryProvider.builder()
.withNodes(bootstrapLocations)
.build())
.build();
cluster1.start().join();
assertEquals("foo", cluster1.getMembershipService().getLocalMember().id().id());
AtomixCluster cluster2 = AtomixCluster.builder()
.withMemberId("bar")
.withAddress("localhost:5001")
.withMembershipProvider(BootstrapDiscoveryProvider.builder()
.withNodes(bootstrapLocations)
.build())
.build();
cluster2.start().join();
cluster1.getCommunicationService().subscribe("test", message -> {
return CompletableFuture.completedFuture("world!");
});
String result = cluster2.getCommunicationService().<String, String>send("test", "Hello!", cluster1.getMembershipService().getLocalMember().id()).join();
System.out.println(result);
if you send a byte array then it’s going to serialize that byte array with a type
not send it raw
so it will copy it, but you should still receive what you sent
Collection<Node> bootstrapLocations = Arrays.asList(
Node.builder().withId("foo").withAddress(Address.from("localhost:5000")).build(),
Node.builder().withId("bar").withAddress(Address.from("localhost:5001")).build(),
Node.builder().withId("baz").withAddress(Address.from("localhost:5002")).build());
AtomixCluster cluster1 = AtomixCluster.builder()
.withMemberId("foo")
.withAddress("localhost:5000")
.withMembershipProvider(BootstrapDiscoveryProvider.builder()
.withNodes(bootstrapLocations)
.build())
.build();
cluster1.start().join();
assertEquals("foo", cluster1.getMembershipService().getLocalMember().id().id());
AtomixCluster cluster2 = AtomixCluster.builder()
.withMemberId("bar")
.withAddress("localhost:5001")
.withMembershipProvider(BootstrapDiscoveryProvider.builder()
.withNodes(bootstrapLocations)
.build())
.build();
cluster2.start().join();
cluster1.getCommunicationService().<byte[], byte[]>subscribe("test", message -> {
System.out.println(new String(message));
return CompletableFuture.completedFuture("world!".getBytes());
});
byte[] result = cluster2.getCommunicationService().<byte[], byte[]>send("test", "Hello!".getBytes(), cluster1.getMembershipService().getLocalMember().id()).join();
System.out.println(new String(result));
both work fine
are you sure that stack trace is not from somewhere else?
nvm
the stack trace is from my client code ,
future.get()
the
future.get() is exactly the at cn.ac.iie.di.atomix3test.atomix.n.Client$T.send(Client.java:150)
I know that’s why I said nvm
and I'm not sure what does
nvm mean...
nevermind
AtomicInteger count = new AtomicInteger();
cluster1.getCommunicationService().subscribe("test_type", (byte[] src) -> {
System.out.println("coming in test_type, addr:" + "unknown" + ", cmd:" + new String(src));
try {
Thread.sleep(1000);
} catch (InterruptedException ex) {
}
return ("return_" + count.getAndIncrement() + "_" + new String(src) + "_from_" + 1).getBytes();
}, new ThreadPoolExecutor(2, 32, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(32), (r, e) -> r.run()));
CompletableFuture<byte[]> future = cluster2.getCommunicationService().send("test_type", "this is a msg to leader".getBytes(), cluster1.getMembershipService().getLocalMember().id());
String response = new String(future.join());
System.out.println(response);
i see
worked for me without an exception
🤔
that exception is not from a
byte[]
it’s from some other type of object being serialized somewhere
when you change
byte[] to a String do you still get the same exception with the same ID?
byte[] is registered with ID 16
I found
String ok
you mean it works with
String?
yes
strange
and not extra information received
something weird with how it’s handling the byte array
can you try my code above?
Collection<Node> bootstrapLocations = Arrays.asList(
Node.builder().withId("foo").withAddress(Address.from("localhost:5000")).build(),
Node.builder().withId("bar").withAddress(Address.from("localhost:5001")).build(),
Node.builder().withId("baz").withAddress(Address.from("localhost:5002")).build());
AtomixCluster cluster1 = AtomixCluster.builder()
.withMemberId("foo")
.withAddress("localhost:5000")
.withMembershipProvider(BootstrapDiscoveryProvider.builder()
.withNodes(bootstrapLocations)
.build())
.build();
cluster1.start().join();
assertEquals("foo", cluster1.getMembershipService().getLocalMember().id().id());
AtomixCluster cluster2 = AtomixCluster.builder()
.withMemberId("bar")
.withAddress("localhost:5001")
.withMembershipProvider(BootstrapDiscoveryProvider.builder()
.withNodes(bootstrapLocations)
.build())
.build();
cluster2.start().join();
cluster1.getCommunicationService().<byte[], byte[]>subscribe("test", message -> {
System.out.println(new String(message));
return CompletableFuture.completedFuture("world!".getBytes());
});
byte[] result = cluster2.getCommunicationService().<byte[], byte[]>send("test", "Hello!".getBytes(), cluster1.getMembershipService().getLocalMember().id()).join();
System.out.println(new String(result));
this works on my machine
I'll try your code
You can remove that junit statement
Rogue codes
I might know the reason
I retried my old code, but the difference is that i removed the
.data dir
and succeeded
data serialized by kyro dependents the running context
but I do set the
StorageLevel to MEMORY.withManagementGroup(RaftPartitionGroup.builder("systen")
.withNumPartitions(1).withMembers("server1", "server2", "server3")
.withStorageLevel(StorageLevel.MEMORY).build())
That’s strange
PS D:\Storage\work\huayan\data_exchange\project\Atomix3Test> ls
目录: D:\Storage\work\huayan\data_exchange\project\Atomix3Test
Mode LastWriteTime Length Name
---- ------------- ------ ----
d----- 2018/8/1 10:10 .data
d----- 2018/4/2 16:42 .projectKnowledge
d----- 2018/4/24 13:49 src
d----- 2018/7/23 10:02 target
-a---- 2018/7/30 16:01 8 .gitignore
-a---- 2018/7/30 16:01 1408 pom.xml
the
.data. dir still exists
Did you have a primitive named
test_type or something?
no
all I did is send and receive msgs
oh, and leader election
in the dir , there is only one file:
PS D:\Storage\work\huayan\data_exchange\project\Atomix3Test\.data\system\partitions\1> ls
目录: D:\Storage\work\huayan\data_exchange\project\Atomix3Test\.data\system\partitions\1
Mode LastWriteTime Length Name
---- ------------- ------ ----
-a---- 2018/8/1 10:19 128 system-partition-1.meta
PS D:\Storage\work\huayan\data_exchange\project\Atomix3Test\.data\system\partitions\1>
as well as in my "data" partition there is a
data-partition-1.meta
That’s just the Raft cluster configuration and term. It doesn’t matter much now, but I’m curious what was sending messages to that handler. Something was sending a message serialized with a different serializer
@kuujo i think DocumentPath.ROOT == "" in tree primitive is not a good idea
Maybe '/' is better
Not sure what you’re talking about
/ is the root path
public static final DocumentPath ROOT = DocumentPath.from(new String[]{""});
That’s just how
/ is represented
In an array
The root node is “”
Err... “” is basically a magic element of the array that represents the root
/. This allows a relative path or child paths to be correctly represented
Could probably change that so it’s not confusing though
Document means a relative path? i think it is absolute path
DocumentPath p = DocumentPath.from("hello", "world");
System.out.print(p.toString()); == > hello/world
System.out.print(p.toString()); == > hello/world
"/hello/world" is more friendly
It’s fine to make that method return an absolute path,
but not all document paths are absolute
but not all document paths are absolute
The implementation will just have to copy the array and pretend a
”” to make it absolute
prepend
My iPad seems to think that’s not a word ^^
Submit a PR
just in tree primitive, if documentpath is relative, get document by DocumentPath is wired
/**
- Returns the child values for this node.
* - @param path path to the node
- @return mapping from a child name to its value
- @throws NoSuchDocumentPathException if the path does not point to a valid node
- @throws IllegalArgumentException if the given path does not start at root {@code /}
*/
Map<String, Versioned<V>> getChildren(DocumentPath path);
I’m not referring to the user API
Internally, paths that are not from the root are used to find nodes
enn, i got your meaning,
So it either has to support both root and child nodes or we have to have two different classes or change the implementation
But the
from(String...) method should be fixed
i got it
+1
Seems like Atomix 3.0.0 will be released around the middle of August
working on a legitimate distribution for the agent
using that thing sucks right now
@kuujo with jigsaw? :wink:
haha no
does Maven even support distributing them
could dump them in the github release section
the artifacts
Yeah not sure
I think so?
must be able to
will find out
i've seen exe's there
should also put them in the github release section too though