class JmxSplitManager implements ConnectorSplitManager {

    @Override
    public ConnectorSplitSource getSplits(
            ConnectorTransactionHandle transaction,
            ConnectorSession session,
            ConnectorTableHandle table,
            SplitSchedulingStrategy splitSchedulingStrategy,
            DynamicFilter dynamicFilter)
    {
        JmxTableHandle tableHandle = (JmxTableHandle) table;

        //TODO is there a better way to get the node column?
        Optional<JmxColumnHandle> nodeColumnHandle = tableHandle.getColumnHandles().stream()
                .filter(jmxColumnHandle -> jmxColumnHandle.getColumnName().equals(NODE_COLUMN_NAME))
                .findFirst();
        checkState(nodeColumnHandle.isPresent(), "Failed to find %s column", NODE_COLUMN_NAME);

        TupleDomain<ColumnHandle> nodeFilter = tableHandle.getNodeFilter();

        List<ConnectorSplit> splits = nodeManager.getAllNodes().stream()
                .filter(node -> {
                    NullableValue value = NullableValue.of(createUnboundedVarcharType(), utf8Slice(node.getNodeIdentifier()));
                    return nodeFilter.overlaps(fromFixedValues(ImmutableMap.of(nodeColumnHandle.get(), value)));
                })
                .map(node -> new JmxSplit(ImmutableList.of(node.getHostAndPort())))
                .collect(toList());

        return new FixedSplitSource(splits);
    }

    @Inject public  JmxSplitManager(NodeManager nodeManager);

}