Lagom cassandra readside table not created

Question

I have created a user repository Lagom readside in my user-impl service, but for some reason the Cassandra users table doesn't get created when running sbt lagom:runAll. I'm not sure what I'm missing here.

My UserRepository class:

@Singleton
public class UserRepository {

    private final CassandraSession session;

    @Inject
    public UserRepository(CassandraSession session, ReadSide readSide) {
        this.session = session;
        readSide.register(PUserEventProcessor.class);
    }

    private static class PUserEventProcessor extends ReadSideProcessor<PUserEvent> {

        private final CassandraSession session;
        private final CassandraReadSide readSide;

        private PreparedStatement insertUserStatement;
        private PreparedStatement updateUserStatement;

        @Inject
        public PUserEventProcessor(CassandraSession session, CassandraReadSide readSide) {
            this.session = session;
            this.readSide = readSide;
        }

        @Override
        public ReadSideHandler<PUserEvent> buildHandler() {
            return readSide.<PUserEvent>builder("pUserEventOffset")
                    .setGlobalPrepare(this::createTables)
                    .setPrepare(tag -> prepareStatements())
                    .setEventHandler(PUserEvent.PUserCreated.class,
                            e -> insertUser(e.getUser()))
                    .setEventHandler(PUserEvent.PUserUpdated.class,
                            e -> updateUser(e.getUser()))
                    .build();
        }

        private void registerCodec(Session session, InstantCodec codec) {
            session.getCluster().getConfiguration().getCodecRegistry().register(codec);
        }

        @Override
        public PSequence<AggregateEventTag<PUserEvent>> aggregateTags() {
            return PUserEvent.TAG.allTags();
        }

        private CompletionStage<Done> createTables() {
            return doAll(
                    session.executeCreateTable(
                            "CREATE TABLE IF NOT EXISTS users (" +
                                    "userId UUID, " +
                                    "email text, " +
                                    "firstName text, " +
                                    "lastName text, " +
                                    "gender text, " +
                                    "PRIMARY KEY (userId) " +
                                    ")"
                    )
            );
        }

        private CompletionStage<Done> prepareStatements() {
            return doAll(
                    session.underlying()
                            .thenAccept(s -> registerCodec(s, InstantCodec.instance))
                            .thenApply(x -> Done.getInstance()),
                    prepareInsertUserStatement(),
                    prepareUpdateUserStatement());
        }


        // Insert users

        private CompletionStage<Done> prepareInsertUserStatement() {
            return session.
                    prepare("INSERT INTO users(" +
                            "userId, " +
                            "email" +
                            "firstName, " +
                            "lastName, " +
                            "gender, " +
                            "refreshToken" +
                            ") " +
                            "VALUES (?, ?, ?, ?, ?, ?)"
                    )
                    .thenApply(accept(s -> insertUserStatement = s));
        }


        private CompletionStage<List<BoundStatement>> insertUser(DbUser user) {
            return completedStatements(
                    insertUserCreator(user)
            );
        }

        private BoundStatement insertUserCreator(DbUser user) {
            return insertUserStatement.bind(
                    user.getUserId(),
                    user.getEmail(),
                    user.getFirstName(),
                    user.getLastName(),
                    user.getGender(),
                    user.getRefreshToken()
            );
        }

        // Update user by userId

        private CompletionStage<Done> prepareUpdateUserStatement() {
            return session.
                    prepare("UPDATE users " +
                            "SET email = ?, " +
                            "   firstName = ?, " +
                            "   lastName = ?, " +
                            "   gender = ?, " +
                            "   refreshToken = ?" +
                            "WHERE userId = ?;"
                    )
                    .thenApply(accept(s -> updateUserStatement = s));
        }


        private CompletionStage<List<BoundStatement>> updateUser(DbUser user) {
            return completedStatements(
                    updateUserCreator(user)
            );
        }

        private BoundStatement updateUserCreator(DbUser user) {
            return updateUserStatement.bind(
                    user.getEmail(),
                    user.getFirstName(),
                    user.getLastName(),
                    user.getGender(),
                    user.getUserId(),
                    user.getRefreshToken()
            );
        }

    }
} 

Added to UserModule.java:

public class UserModule extends AbstractModule implements ServiceGuiceSupport {

    @Override
    protected void configure() {
        bindService(UserService.class, UserServiceImpl.class);
        bind(UserRepository.class);
    }
}

application.conf:

play.modules.enabled += UserModule
lagom.persistence.ask-timeout = 1000s

user.cassandra.keyspace = user

cassandra-journal.keyspace = ${user.cassandra.keyspace}
cassandra-snapshot-store.keyspace = ${user.cassandra.keyspace}
lagom.persistence.read-side.cassandra.keyspace = ${user.cassandra.keyspace}

My setup is standard:

The 'user' table is created, the message table even stores events.
Any help on why the 'users' table does not appear, or how to troubleshoot?

This sort of driving me nuts.
No errors get thrown when starting the services, checked the lagom-auction example and the keyspaces are available there. I don't see any difference in the code...

Edit: More attempts to get my table: - sbt clean / manually deleting the embedded cassandra - Upgrade lagom-sbt-plugin to 1.4.8

Help or even tips much appreciated. Cannot do much without a readside, lol.

Answer 1

You need to bind the UserRepository as an eager singleton. This forces the initialization to not be lazy. You'll notice that it doesn't matter in prod, but it does for development (which is likely why you're seeing this problem in dev).

For an example, see the online auction.